Pydantic AI Documentation Server

import logging from fastmcp import FastMCP from typing import List, Optional from pathlib import Path # Import actual models from .models import ( ParsedDocument, TopicItem, StatusResponse, ChangelogFile, ChangelogContent, ) # Import repository manager functions from .repository_manager import ( clone_or_pull_repository, get_repo_path, get_current_commit_hash, ) # Import parser functions from .parser import parse_markdown_file, get_pydantic_docs_path # Search index functionality (Whoosh-based) has been removed from this server. logger = logging.getLogger(__name__) app = FastMCP() @app.tool() async def update_documentation(force_clone: bool = False) -> StatusResponse: """Clones/updates the Pydantic repo, parses docs, and rebuilds the search index.""" logger.info(f"update_documentation called with force_clone={force_clone}") try: repo_update_result = clone_or_pull_repository(force_clone) logger.info(f"Repository update result: {repo_update_result}") if repo_update_result.get("status") == "error": return StatusResponse( status="error", message=repo_update_result.get("message", "Repository update failed"), data=repo_update_result, ) commit_hash = get_current_commit_hash() return StatusResponse( status="success", message=f"Documentation updated successfully from commit {commit_hash}.", data={"commit_hash": commit_hash}, ) except Exception as e: logger.error(f"Error in update_documentation: {e}", exc_info=True) return StatusResponse( status="error", message=str(e), data={"error_details": str(e)} ) @app.tool() async def get_document_by_path(path: str) -> Optional[ParsedDocument]: """ Retrieves a specific document by its path relative to the Pydantic documentation root (e.g., 'usage/models.md'). Returns the ParsedDocument if found, otherwise None. """ logger.info(f"Attempting to get document by path: {path}") try: pydantic_docs_root = get_pydantic_docs_path() full_file_path = (pydantic_docs_root / path).resolve() if not str(full_file_path).startswith(str(pydantic_docs_root.resolve())): logger.warning( f"Path traversal attempt or invalid path for get_document_by_path: {path}" ) return None if not full_file_path.is_file(): logger.info( f"Document not found or not a file at path: {path} (resolved: {full_file_path})" ) return None document = parse_markdown_file( file_path=full_file_path, docs_base_dir=pydantic_docs_root ) if document: logger.info(f"Successfully retrieved document: {document.path}") return document else: logger.warning( f"Failed to parse document at path: {path} (resolved: {full_file_path})" ) return None except Exception as e: logger.error( f"Error in get_document_by_path for path '{path}': {e}", exc_info=True ) return None @app.tool() async def list_topics(path: Optional[str] = None) -> List[TopicItem]: """ Lists files and directories non-recursively within the Pydantic documentation. The path is relative to the 'docs/' directory in the cloned Pydantic repository. """ logger.info(f"Listing topics for path: {path if path else 'root'}") try: repo_root = get_repo_path() docs_root = repo_root / "docs" if not docs_root.exists() or not docs_root.is_dir(): logger.error(f"Pydantic docs directory not found at {docs_root}") return [] current_path = docs_root if path: target_path = (docs_root / path).resolve(strict=False) if not str(target_path).startswith(str(docs_root.resolve())): logger.warning(f"Path traversal attempt or invalid path: {path}") return [] if not target_path.exists() or not target_path.is_dir(): logger.info( f"Specified path does not exist or is not a directory: {target_path}" ) return [] current_path = target_path items: List[TopicItem] = [] for item in current_path.iterdir(): if item.name.startswith("."): continue item_relative_path = item.relative_to(docs_root).as_posix() items.append( TopicItem( name=item.name, path=item_relative_path, is_directory=item.is_dir() ) ) items.sort(key=lambda x: (not x.is_directory, x.name.lower())) return items except FileNotFoundError: logger.info(f"Path not found during list_topics: {path}") return [] except Exception as e: logger.error(f"Error listing topics for path '{path}': {e}", exc_info=True) return [] @app.tool() async def list_available_changelogs() -> List[ChangelogFile]: """ Lists all available changelog files found in the Pydantic-AI documentation repository (e.g., in 'docs/history/'). """ logger.info("list_available_changelogs called") changelogs: List[ChangelogFile] = [] try: repo_root = get_repo_path() history_dir = repo_root / "docs" / "history" if not history_dir.is_dir(): logger.warning(f"Changelog directory not found: {history_dir}") return [] for file_path in history_dir.glob("*.md"): if file_path.is_file(): version_name = file_path.stem relative_path = file_path.relative_to(history_dir).as_posix() changelogs.append( ChangelogFile( name=file_path.name, path=f"history/{relative_path}", version=version_name, ) ) changelogs.sort( key=lambda cf: [ int(part) if part.isdigit() else part for part in cf.version.split(".") ], reverse=True, ) logger.info(f"Found {len(changelogs)} changelog files.") except Exception as e: logger.error(f"Error listing available changelogs: {e}", exc_info=True) return [] return changelogs @app.tool() async def get_changelog_content(path: str) -> Optional[ChangelogContent]: """ Retrieves the parsed content of a specific changelog file. The path should be relative to the Pydantic-AI documentation root's 'docs' directory (e.g., 'history/0.1.0.md'). """ logger.info(f"get_changelog_content called for path: {path}") try: repo_root = get_repo_path() docs_dir = repo_root / "docs" changelog_file_path = (docs_dir / path).resolve() if not str(changelog_file_path).startswith(str(docs_dir.resolve())): logger.warning( f"Path traversal attempt or invalid path for get_changelog_content: {path}" ) return None if not changelog_file_path.is_file(): logger.info( f"Changelog file not found or not a file at path: {path} (resolved: {changelog_file_path})" ) return None content = changelog_file_path.read_text(encoding="utf-8") version = Path(path).stem return ChangelogContent( path=path, version=version, content=content, release_date=None, ) except Exception as e: logger.error( f"Error in get_changelog_content for path '{path}': {e}", exc_info=True ) return None def main_server_logic(): logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", ) logger.info("Initializing Pydantic Docs Server...") app.run(transport="stdio") # This check is not strictly necessary if __main__.py calls main_server_logic directly, # but good practice if server.py might also be run directly. if __name__ == "__main__": main_server_logic()