-
Notifications
You must be signed in to change notification settings - Fork 379
Add documentation tools #39
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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") | ||
Comment on lines
+29
to
+35
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @shiraayal-tadata Second, all these options should be explained in the README too, not just here. |
||
|
||
# Run the server if this file is executed directly | ||
if __name__ == "__main__": | ||
import uvicorn | ||
|
||
uvicorn.run(app, host="127.0.0.1", port=8000) |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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): | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @shiraayal-tadata is file a Path? a str? missing typing. |
||
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 |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,3 @@ | ||
import logging | ||
|
||
logger = logging.getLogger("fastapi_mcp") |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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 |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@shiraayal-tadata I would expect documentation tools to be a parameter and not a separate function we need to call.
The
add_mcp_server()
function takes a FastAPI app and mutates it, and that makes sense because we are not in control of the FastAPI instance.But we are in control of our own MCP server instance, so there's no need for another mutating function that takes a server instance. We can just add it as a feature to the existing
add_mcp_server()
function.A clearer syntax might be something like this:
Here, "documentation_tools" is a parameter expecting a set of tools in some format, and "create_documentation_tools" is not mutating any fastMCP instance. It's just a helper function that returns tools.