diff --git a/README.md b/README.md index f52e046..301106b 100644 --- a/README.md +++ b/README.md @@ -83,6 +83,19 @@ async def get_server_time() -> str: return datetime.now().isoformat() ``` +You can also auto-create a documentation tool, that will easily expose your docs to the AI agent. Refer to the [detailed example](examples/mcp_server_with_docs.py). + +```python +from fastapi import FastAPI +from fastapi_mcp import add_mcp_server +from fastapi_mcp.documentation_tools import create_documentation_tools + +app = FastAPI() +mcp_server = add_mcp_server(app) + +create_documentation_tools(mcp_server, "README.md") +``` + ## Examples See the [examples](examples) directory for complete examples. diff --git a/examples/mcp_server_with_docs.py b/examples/mcp_server_with_docs.py new file mode 100644 index 0000000..e3cc36e --- /dev/null +++ b/examples/mcp_server_with_docs.py @@ -0,0 +1,41 @@ +from fastapi import FastAPI +from fastapi_mcp import add_mcp_server +from fastapi_mcp.documentation_tools import create_documentation_tools + +# Create a very basic FastAPI app +app = FastAPI( + title="Basic API", + description="A basic API with integrated MCP server", + version="0.1.0", +) + +@app.get("/") +async def root(): + return {"message": "Hello World"} + +@app.get("/greet/{name}") +async def greet(name: str): + return {"message": f"Hello, {name}!"} + +# Add MCP server to the FastAPI app +mcp_server = add_mcp_server( + app, + mount_path="/mcp", + name="Basic API MCP", + description="MCP server for the Basic API", + base_url="http://localhost:8000", +) + +# Create documentation tools for the MCP server +# Based on a single file +create_documentation_tools(mcp_server, "README.md") +# Based on a list of files +# create_documentation_tools(mcp_server, ["README.md", "README2.md", "llms.txt"]) +# Based on a directory +# create_documentation_tools(mcp_server, "docs") + +# Run the server if this file is executed directly +if __name__ == "__main__": + import uvicorn + + uvicorn.run(app, host="127.0.0.1", port=8000) diff --git a/fastapi_mcp/documentation_tools.py b/fastapi_mcp/documentation_tools.py new file mode 100644 index 0000000..e6fd88c --- /dev/null +++ b/fastapi_mcp/documentation_tools.py @@ -0,0 +1,142 @@ +from pathlib import Path +from mcp.server.fastmcp import FastMCP +from typing import Dict, Any, Union, List, Optional +from dataclasses import dataclass, asdict +from .logger import logger + +@dataclass +class ContentItem: + type: str = "text" + text: str = "" + source: Optional[str] = None + +@dataclass +class DocumentationResponse: + docs: List[ContentItem] + + @classmethod + def error(cls) -> "DocumentationResponse": + return cls(docs=[ContentItem(text="Error getting documentation")]) + + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + +def _get_file_priority(file): + name = file.name.lower() + if name == 'llms.txt': + return 0 # Highest priority + elif name.startswith('readme'): + return 1 # Second priority + else: + return 2 # Lowest priority + +def _prioritize_files(files: List[Path]) -> List[Path]: + """ + Return only the highest priority files available. + If priority 0 files exist, return only those. + If not, return priority 1 files. + If neither exist, return priority 2 files. + + Args: + files: List of files to prioritize + + Returns: + List of files with the highest available priority + """ + if not files: + return [] + + # Group files by priority + priority_groups: Dict[int, List[Path]] = {0: [], 1: [], 2: []} + for file in files: + priority = _get_file_priority(file) + priority_groups[priority].append(file) + + # Return the highest priority group that has files + for priority in [0, 1, 2]: + if priority_groups[priority]: + logger.info(f"Found {len(priority_groups[priority])} files with priority {priority}") + return priority_groups[priority] + + return [] # Should never reach here if files is non-empty + +def _collect_documentation_files(docs_path: Union[str, List[str]]) -> List[Path]: + """ + Collect documentation files from the given path(s). + + Args: + docs_path: Can be one of: + - A single path to a documentation file (str) + - A list of paths to documentation files (List[str]) + - A path to a directory containing documentation files (str) + + Returns: + List of Path objects representing the documentation files + """ + # Handle list of file paths + if isinstance(docs_path, list): + files = [] + for path_str in docs_path: + path = Path(path_str) + if path.exists(): + files.append(path) + return files + + path = Path(docs_path) + + # If it's a directory, find all markdown files and llms.txt files + if path.is_dir(): + md_files = list(path.rglob('*.md')) + llms_txt_files = list(path.rglob('llms.txt')) + return _prioritize_files(llms_txt_files + md_files) + + # If it's a single file, return it + return [path] + + +def _read_documentation_files(files: List[Path]) -> List[tuple[str, str]]: + all_content = [] + for file in files: + try: + with open(file, 'r', encoding='utf-8') as f: + content = f.read() + all_content.append((content, str(file))) + except Exception as e: + logger.error(f"Error reading documentation file {file}: {e}") + return all_content + + +def create_documentation_tools(mcp_server: FastMCP, docs_path: Union[str, List[str]]) -> None: + """ + Create MCP tools that serve the documentation to an AI Agent and allows searching through documentation files. + + Args: + mcp_server: The MCP server to add the tool to + docs_path: Can be one of: + - A single path to a documentation file (str) + - A list of paths to documentation files (List[str]) + - A path to a directory containing documentation files (str) + """ + @mcp_server.tool() + async def fetch_documentation() -> Dict[str, Any]: + """ + Fetch the contents of documentation files. + + Returns: + A dictionary containing the file content and metadata + """ + files = _collect_documentation_files(docs_path) + if not files: + logger.error(f"No documentation files found at {docs_path}") + return DocumentationResponse.error().to_dict() + + all_content = _read_documentation_files(files) + if not all_content: + logger.error(f"No content found in documentation files at {docs_path}") + return DocumentationResponse.error().to_dict() + + return DocumentationResponse( + docs=[ContentItem(text=content, source=source) for content, source in all_content], + ).to_dict() + + # TODO: Add tool to search documentation \ No newline at end of file diff --git a/fastapi_mcp/http_tools.py b/fastapi_mcp/http_tools.py index 233ae96..c18c182 100644 --- a/fastapi_mcp/http_tools.py +++ b/fastapi_mcp/http_tools.py @@ -5,7 +5,6 @@ """ import json -import logging from typing import Any, Callable, Dict, List, Optional import httpx @@ -13,6 +12,8 @@ from fastapi.openapi.utils import get_openapi from mcp.server.fastmcp import FastMCP +from .logger import logger + from .openapi_utils import ( clean_schema_for_display, generate_example_from_schema, @@ -22,7 +23,6 @@ PYTHON_TYPE_IMPORTS, ) -logger = logging.getLogger("fastapi_mcp") def create_mcp_tools_from_openapi( diff --git a/fastapi_mcp/logger.py b/fastapi_mcp/logger.py new file mode 100644 index 0000000..345ca94 --- /dev/null +++ b/fastapi_mcp/logger.py @@ -0,0 +1,3 @@ +import logging + +logger = logging.getLogger("fastapi_mcp") diff --git a/tests/test_documentation_tools.py b/tests/test_documentation_tools.py new file mode 100644 index 0000000..193ff5f --- /dev/null +++ b/tests/test_documentation_tools.py @@ -0,0 +1,146 @@ +import pytest +import json +from pathlib import Path +from fastapi_mcp.documentation_tools import ( + _get_file_priority, + _prioritize_files, + _collect_documentation_files, + _read_documentation_files, + ContentItem, + DocumentationResponse, + create_documentation_tools, +) +from mcp.server.fastmcp import FastMCP + + +def test_get_file_priority(): + """Test the file priority system""" + llms_file = Path("llms.txt") + readme_file = Path("README.md") + other_file = Path("documentation.md") + + assert _get_file_priority(llms_file) == 0 # Highest priority + assert _get_file_priority(readme_file) == 1 # Second priority + assert _get_file_priority(other_file) == 2 # Lowest priority + + +def test_prioritize_files(): + """Test the file prioritization logic""" + llms_file = Path("llms.txt") + readme_file = Path("README.md") + other_file = Path("documentation.md") + + # Test with mixed priority files + files = [other_file, readme_file, llms_file] + prioritized = _prioritize_files(files) + assert len(prioritized) == 1 + assert prioritized[0] == llms_file + + # Test with only readme and other files + files = [other_file, readme_file] + prioritized = _prioritize_files(files) + assert len(prioritized) == 1 + assert prioritized[0] == readme_file + + # Test with only other files + files = [other_file] + prioritized = _prioritize_files(files) + assert len(prioritized) == 1 + assert prioritized[0] == other_file + + # Test with empty list + assert _prioritize_files([]) == [] + + +def test_collect_documentation_files(tmp_path): + """Test documentation file collection using a real temporary filesystem""" + # Create a docs directory with test files + docs_dir = tmp_path / "docs" + docs_dir.mkdir() + + # Create test files + (docs_dir / "llms.txt").write_text("LLM documentation") + (docs_dir / "README.md").write_text("README content") + (docs_dir / "other.md").write_text("Other documentation") + + # Test directory path - should return only llms.txt due to priority + files = _collect_documentation_files(str(docs_dir)) + assert len(files) == 1 + assert files[0].name == "llms.txt" + + # Test single file path + single_file = docs_dir / "README.md" + files = _collect_documentation_files(str(single_file)) + assert len(files) == 1 + assert files[0].name == "README.md" + + # Test list of files + file_list = [ + str(docs_dir / "README.md"), + str(docs_dir / "other.md") + ] + files = _collect_documentation_files(file_list) + assert len(files) == 2 + assert {f.name for f in files} == {"README.md", "other.md"} + + +def test_read_documentation_files(tmp_path): + """Test reading documentation files from filesystem""" + # Create test files with content + file1 = tmp_path / "test1.md" + file2 = tmp_path / "test2.md" + + file1.write_text("Content 1") + file2.write_text("Content 2") + + files = [file1, file2] + content = _read_documentation_files(files) + + assert len(content) == 2 + assert content[0][0] == "Content 1" + assert content[1][0] == "Content 2" + assert content[0][1] == str(file1) + assert content[1][1] == str(file2) + + +@pytest.mark.asyncio +async def test_fetch_documentation(tmp_path): + """Test the fetch_documentation tool with real files""" + # Create test documentation + docs_dir = tmp_path / "docs" + docs_dir.mkdir() + test_content = "Test documentation content" + (docs_dir / "llms.txt").write_text(test_content) + + # Create and test the tool + mcp = FastMCP() + create_documentation_tools(mcp, str(docs_dir)) + + # Test the tool + result = await mcp.call_tool("fetch_documentation", {}) + assert len(result) == 1 + result_dict = json.loads(result[0].text) + assert "docs" in result_dict + docs = result_dict["docs"] + assert len(docs) == 1 + assert docs[0]["text"] == test_content + assert docs[0]["source"] == str(docs_dir / "llms.txt") + + + +def test_documentation_response(): + """Test DocumentationResponse class""" + # Test normal response + docs = [ContentItem(text="test", source="test.md")] + response = DocumentationResponse(docs=docs) + assert response.docs == docs + + # Test error response + error_response = DocumentationResponse.error() + assert len(error_response.docs) == 1 + assert error_response.docs[0].text == "Error getting documentation" + + # Test to_dict + response_dict = response.to_dict() + assert isinstance(response_dict, dict) + assert 'docs' in response_dict \ No newline at end of file