YouTube Search MCP Server

# FastMCP Framework - Comprehensive Documentation ## Overview FastMCP is a Python framework for building and interacting with Model Context Protocol (MCP) servers. It provides a "USB-C port for AI" - a standardized way to expose data and functionality to AI applications through a high-level, Pythonic interface. **Source Code**: https://github.com/jlowin/fastmcp ## Core Concepts ### What is MCP? The Model Context Protocol (MCP) is an open standard that enables AI applications to securely connect to data sources and tools. It provides a structured way for AI systems to: - Access external resources and data - Execute tools and functions - Interact with prompts and templates - Maintain secure, authenticated connections ### FastMCP Architecture FastMCP consists of two main components: 1. **Server**: Exposes tools, resources, and prompts to AI clients 2. **Client**: Connects to and interacts with MCP servers ## Getting Started ### Installation ```bash pip install fastmcp ``` ### Basic Server Example ```python from fastmcp import FastMCP # Create a new MCP server mcp = FastMCP("Demo Server 🚀") @mcp.tool def add(a: int, b: int) -> int: """Add two numbers together""" return a + b @mcp.tool def greet(name: str) -> str: """Greet someone by name""" return f"Hello, {name}!" # Run the server if __name__ == "__main__": mcp.run() ``` ## Core Features ### 1. Tools Tools are executable functions that AI clients can call. They're defined using the `@mcp.tool` decorator. ```python @mcp.tool def search_database(query: str, limit: int = 10) -> list: """Search the database for matching records""" # Implementation here return results @mcp.tool def process_file(filepath: str, operation: str) -> dict: """Process a file with the specified operation""" # Implementation here return {"status": "success", "result": processed_data} ``` ### 2. Resources Resources provide access to data sources and content. They can be static or templated. ```python @mcp.resource("user-data") def get_user_data() -> str: """Get current user data""" return json.dumps(user_data) @mcp.resource("file://{path}") def get_file_content(path: str) -> str: """Get content of a file""" with open(path, 'r') as f: return f.read() ``` ### 3. Prompts Prompts are reusable templates that can be parameterized for different use cases. ```python @mcp.prompt("analyze-code") def analyze_code_prompt(language: str, code: str) -> str: """Generate a prompt for code analysis""" return f""" Analyze this {language} code and provide feedback: ```{language} {code} ``` Please focus on: - Code quality - Performance - Security issues - Best practices """ ``` ## Server Components ### FastMCP Server Class The main server class that handles MCP protocol communication: ```python from fastmcp.server import FastMCP # Create server with custom configuration mcp = FastMCP( name="My Server", version="1.0.0", description="A sample MCP server" ) ``` ### Authentication FastMCP supports multiple authentication methods: #### Bearer Token Authentication ```python from fastmcp.server.auth import BearerTokenAuth # Environment variable authentication auth = BearerTokenAuth.from_env("MY_API_TOKEN") mcp = FastMCP("Secure Server", auth=auth) ``` #### OAuth Authentication ```python from fastmcp.server.auth import OAuthAuth oauth = OAuthAuth( client_id="your-client-id", client_secret="your-client-secret", authorization_url="https://auth.example.com/oauth/authorize", token_url="https://auth.example.com/oauth/token" ) mcp = FastMCP("OAuth Server", auth=oauth) ``` ### Middleware Add cross-cutting functionality to your server: ```python from fastmcp.server.middleware import LoggingMiddleware, TimingMiddleware mcp = FastMCP("Middleware Server") mcp.add_middleware(LoggingMiddleware()) mcp.add_middleware(TimingMiddleware()) ``` ### Context and Dependencies Access MCP capabilities within your functions: ```python from fastmcp.server.context import MCPContext @mcp.tool def long_running_task(ctx: MCPContext) -> str: """A task that reports progress""" ctx.progress("Starting task...", 0.0) # Do some work ctx.progress("Half way done...", 0.5) # More work ctx.progress("Almost finished...", 0.9) # Final work ctx.progress("Complete!", 1.0) return "Task completed successfully" ``` ## Client Components ### FastMCP Client Connect to and interact with MCP servers: ```python from fastmcp.client import FastMCPClient # Connect to a server client = FastMCPClient("stdio://path/to/server") # List available tools tools = client.list_tools() # Call a tool result = client.call_tool("add", {"a": 5, "b": 3}) # Get resources resources = client.list_resources() content = client.get_resource("user-data") # Use prompts prompts = client.list_prompts() prompt = client.get_prompt("analyze-code", {"language": "python", "code": "print('hello')"}) ``` ### Transport Protocols FastMCP supports multiple transport methods: #### STDIO Transport ```python from fastmcp.client.transports import StdioTransport transport = StdioTransport("python", "server.py") client = FastMCPClient(transport) ``` #### HTTP Transport ```python from fastmcp.client.transports import HttpTransport transport = HttpTransport("http://localhost:8000") client = FastMCPClient(transport) ``` #### Server-Sent Events (SSE) ```python from fastmcp.client.transports import SSETransport transport = SSETransport("http://localhost:8000/sse") client = FastMCPClient(transport) ``` ## Advanced Features ### Server Composition Combine multiple servers into a single application: ```python from fastmcp.server.composition import mount_server # Main server main_server = FastMCP("Main Server") # Sub-servers auth_server = FastMCP("Auth Server") data_server = FastMCP("Data Server") # Mount sub-servers main_server.mount("/auth", auth_server) main_server.mount("/data", data_server) ``` ### Tool Transformation Create enhanced tool variants: ```python from fastmcp.tools.tool_transform import transform_tool @transform_tool( name="enhanced_search", description="Search with additional filtering", schema_updates={"properties": {"filter": {"type": "string"}}} ) def search_with_filter(original_func, filter: str = None, **kwargs): """Enhanced search with filtering""" results = original_func(**kwargs) if filter: results = apply_filter(results, filter) return results ``` ### Caching Implement caching for performance: ```python from fastmcp.utilities.cache import cache @mcp.tool @cache(ttl=300) # Cache for 5 minutes def expensive_operation(data: str) -> str: """An expensive operation that benefits from caching""" # Expensive computation here return processed_result ``` ### Testing Test your MCP servers effectively: ```python from fastmcp.utilities.tests import MCPTestClient def test_my_server(): client = MCPTestClient(mcp) # Test tool execution result = client.call_tool("add", {"a": 2, "b": 3}) assert result == 5 # Test resource access content = client.get_resource("user-data") assert content is not None ``` ## Integration Examples ### FastAPI Integration ```python from fastapi import FastAPI from fastmcp.integrations.fastapi import FastMCPIntegration app = FastAPI() mcp = FastMCP("FastAPI Server") # Add MCP integration FastMCPIntegration(app, mcp) @mcp.tool def api_tool(message: str) -> str: return f"API says: {message}" ``` ### OpenAPI Integration ```python from fastmcp.integrations.openapi import create_mcp_from_openapi # Generate MCP server from OpenAPI spec mcp = create_mcp_from_openapi( openapi_url="https://api.example.com/openapi.json", server_name="API Server" ) ``` ## Deployment ### Running Your Server ```python # Development mode mcp.run(debug=True) # Production mode with specific transport mcp.run(transport="http", host="0.0.0.0", port=8000) # STDIO mode (for direct integration) mcp.run(transport="stdio") ``` ### Configuration ```python from fastmcp.settings import MCPSettings settings = MCPSettings( server_name="Production Server", max_connections=100, timeout=30, log_level="INFO" ) mcp = FastMCP(settings=settings) ``` ## Best Practices 1. **Use Type Hints**: Always provide type hints for better IDE support and validation 2. **Error Handling**: Implement proper error handling in your tools and resources 3. **Security**: Use authentication for production servers 4. **Performance**: Implement caching for expensive operations 5. **Testing**: Write comprehensive tests for your MCP components 6. **Documentation**: Provide clear docstrings for all tools, resources, and prompts ## Common Patterns ### Database Integration ```python import sqlite3 from fastmcp.server.context import MCPContext @mcp.tool def query_database(ctx: MCPContext, query: str) -> list: """Execute a database query""" try: conn = sqlite3.connect("database.db") cursor = conn.cursor() cursor.execute(query) results = cursor.fetchall() conn.close() return results except Exception as e: ctx.log_error(f"Database query failed: {e}") raise ``` ### File Operations ```python import os from pathlib import Path @mcp.resource("files://{path}") def get_file_listing(path: str) -> list: """Get listing of files in a directory""" directory = Path(path) if not directory.exists(): return [] return [ { "name": f.name, "type": "file" if f.is_file() else "directory", "size": f.stat().st_size if f.is_file() else 0 } for f in directory.iterdir() ] ``` ### External API Integration ```python import requests from fastmcp.utilities.cache import cache @mcp.tool @cache(ttl=600) # Cache for 10 minutes def fetch_weather(city: str) -> dict: """Fetch weather data for a city""" response = requests.get(f"https://api.weather.com/v1/current?q={city}") return response.json() ``` ## Troubleshooting ### Common Issues 1. **Connection Errors**: Check transport configuration and network connectivity 2. **Authentication Failures**: Verify tokens and credentials 3. **Tool Execution Errors**: Ensure proper error handling and type validation 4. **Performance Issues**: Implement caching and optimize database queries ### Debugging ```python import logging from fastmcp.utilities.logging import setup_logging # Enable debug logging setup_logging(level=logging.DEBUG) # Run server in debug mode mcp.run(debug=True) ``` ## Community and Resources - **GitHub Repository**: https://github.com/jlowin/fastmcp - **Documentation**: https://gofastmcp.com/ - **Community Showcase**: https://gofastmcp.com/community/showcase.md - **Contributing**: See repository for contribution guidelines ## License and Support FastMCP is open source and actively maintained. Check the GitHub repository for the latest updates, issues, and contribution guidelines.