MCP Web Tools Server

  • docs
# Building MCP Servers with Python This guide provides a comprehensive walkthrough for building Model Context Protocol (MCP) servers using Python. We'll cover everything from basic setup to advanced techniques, with practical examples and best practices. ## Prerequisites Before starting, ensure you have: - Python 3.10 or higher installed - Basic knowledge of Python and async programming - Understanding of MCP core concepts (tools, resources, prompts) - A development environment with your preferred code editor ## Setting Up Your Environment ### Installation Start by creating a virtual environment and installing the MCP package: ```bash # Create a virtual environment python -m venv venv source venv/bin/activate # On Windows: venv\Scripts\activate # Install MCP pip install mcp ``` Alternatively, if you're using [uv](https://github.com/astral-sh/uv) for package management: ```bash # Create a virtual environment uv venv source .venv/bin/activate # On Windows: .venv\Scripts\activate # Install MCP uv pip install mcp ``` ### Project Structure A well-organized MCP server project typically follows this structure: ``` my-mcp-server/ ├── requirements.txt ├── server.py ├── tools/ │ ├── __init__.py │ ├── tool_module1.py │ └── tool_module2.py ├── resources/ │ ├── __init__.py │ └── resource_modules.py └── prompts/ ├── __init__.py └── prompt_modules.py ``` This modular structure keeps your code organized and makes it easier to add new functionality over time. ## Creating Your First MCP Server ### Basic Server Structure Let's create a simple MCP server with a "hello world" tool: ```python # server.py from mcp.server.fastmcp import FastMCP # Create a server mcp = FastMCP("HelloWorld") @mcp.tool() def hello(name: str = "World") -> str: """ Say hello to a name. Args: name: The name to greet (default: "World") Returns: A greeting message """ return f"Hello, {name}!" if __name__ == "__main__": # Run the server mcp.run() ``` This basic server: 1. Creates a FastMCP server named "HelloWorld" 2. Defines a simple tool called "hello" that takes a name parameter 3. Runs the server using the default stdio transport ### Running Your Server To run your server: ```bash python server.py ``` The server will start and wait for connections on the standard input/output streams. ### FastMCP vs. Low-Level API The MCP Python SDK provides two ways to create servers: 1. **FastMCP**: A high-level API that simplifies server creation through decorators 2. **Low-Level API**: Provides more control but requires more boilerplate code Most developers should start with FastMCP, as it handles many details automatically. ## Implementing Tools Tools are the most common primitive in MCP servers. They allow LLMs to perform actions and retrieve information. ### Basic Tool Example Here's how to implement a simple calculator tool: ```python @mcp.tool() def calculate(operation: str, a: float, b: float) -> float: """ Perform basic arithmetic operations. Args: operation: The operation to perform (add, subtract, multiply, divide) a: First number b: Second number Returns: The result of the operation """ if operation == "add": return a + b elif operation == "subtract": return a - b elif operation == "multiply": return a * b elif operation == "divide": if b == 0: raise ValueError("Cannot divide by zero") return a / b else: raise ValueError(f"Unknown operation: {operation}") ``` ### Asynchronous Tools For operations that involve I/O or might take time, use async tools: ```python @mcp.tool() async def fetch_weather(city: str) -> str: """ Fetch weather information for a city. Args: city: The city name Returns: Weather information """ async with httpx.AsyncClient() as client: response = await client.get(f"https://weather-api.example.com/{city}") data = response.json() return f"Temperature: {data['temp']}°C, Conditions: {data['conditions']}" ``` ### Tool Parameters Tools can have: - Required parameters - Optional parameters with defaults - Type hints that are used to generate schema - Docstrings that provide descriptions ```python @mcp.tool() def search_database( query: str, limit: int = 10, offset: int = 0, sort_by: str = "relevance" ) -> list: """ Search the database for records matching the query. Args: query: The search query string limit: Maximum number of results to return (default: 10) offset: Number of results to skip (default: 0) sort_by: Field to sort results by (default: "relevance") Returns: List of matching records """ # Implementation details... return results ``` ### Error Handling in Tools Proper error handling is essential for robust tools: ```python @mcp.tool() def divide(a: float, b: float) -> float: """ Divide two numbers. Args: a: Numerator b: Denominator Returns: The division result Raises: ValueError: If attempting to divide by zero """ try: if b == 0: raise ValueError("Cannot divide by zero") return a / b except Exception as e: # Log the error for debugging logging.error(f"Error in divide tool: {str(e)}") # Re-raise with a user-friendly message raise ValueError(f"Division failed: {str(e)}") ``` ### Grouping Related Tools For complex servers, organize related tools into modules: ```python # tools/math_tools.py def register_math_tools(mcp): @mcp.tool() def add(a: float, b: float) -> float: """Add two numbers.""" return a + b @mcp.tool() def subtract(a: float, b: float) -> float: """Subtract b from a.""" return a - b # More math tools... # server.py from tools.math_tools import register_math_tools mcp = FastMCP("MathServer") register_math_tools(mcp) ``` ## Implementing Resources Resources provide data to LLMs through URI-based access patterns. ### Basic Resource Example Here's a simple file resource: ```python @mcp.resource("file://{path}") async def get_file(path: str) -> str: """ Get the content of a file. Args: path: Path to the file Returns: The file content """ try: async with aiofiles.open(path, "r") as f: return await f.read() except Exception as e: raise ValueError(f"Failed to read file: {str(e)}") ``` ### Dynamic Resources Resources can be dynamic and parameterized: ```python @mcp.resource("database://{table}/{id}") async def get_database_record(table: str, id: str) -> str: """ Get a record from the database. Args: table: The table name id: The record ID Returns: The record data """ # Implementation details... return json.dumps(record) ``` ### Resource Metadata Resources can include metadata: ```python @mcp.resource("api://{endpoint}") async def get_api_data(endpoint: str) -> tuple: """ Get data from an API endpoint. Args: endpoint: The API endpoint path Returns: A tuple of (content, mime_type) """ async with httpx.AsyncClient() as client: response = await client.get(f"https://api.example.com/{endpoint}") return response.text, response.headers.get("content-type", "text/plain") ``` ### Binary Resources Resources can return binary data: ```python from mcp.server.fastmcp import Image @mcp.resource("image://{path}") async def get_image(path: str) -> Image: """ Get an image file. Args: path: Path to the image Returns: The image data """ with open(path, "rb") as f: data = f.read() return Image(data=data, format=path.split(".")[-1]) ``` ## Implementing Prompts Prompts are templates that help LLMs interact with your server effectively. ### Basic Prompt Example Here's a simple query prompt: ```python @mcp.prompt() def search_query(query: str) -> str: """ Create a search query prompt. Args: query: The search query Returns: Formatted search query prompt """ return f""" Please search for information about: {query} Focus on the most relevant and up-to-date information. """ ``` ### Multi-Message Prompts Prompts can include multiple messages: ```python from mcp.types import UserMessage, AssistantMessage @mcp.prompt() def debug_error(error: str) -> list: """ Create a debugging conversation. Args: error: The error message Returns: A list of messages """ return [ UserMessage(f"I'm getting this error: {error}"), AssistantMessage("Let me help debug that. What have you tried so far?") ] ``` ## Transport Options MCP supports different transport mechanisms for communication between clients and servers. ### STDIO Transport (Default) The default transport uses standard input/output streams: ```python if __name__ == "__main__": mcp.run(transport="stdio") ``` This is ideal for local processes and command-line tools. ### SSE Transport Server-Sent Events (SSE) transport is used for web applications: ```python if __name__ == "__main__": mcp.run(transport="sse", host="localhost", port=5000) ``` This starts an HTTP server that accepts MCP connections through SSE. ## Context and Lifespan ### Using Context The `Context` object provides access to the current request context: ```python from mcp.server.fastmcp import Context @mcp.tool() async def log_message(message: str, ctx: Context) -> str: """ Log a message and return a confirmation. Args: message: The message to log ctx: The request context Returns: Confirmation message """ ctx.info(f"User logged: {message}") return f"Message logged: {message}" ``` ### Progress Reporting For long-running tools, report progress: ```python @mcp.tool() async def process_files(files: list[str], ctx: Context) -> str: """ Process multiple files with progress tracking. Args: files: List of file paths ctx: The request context Returns: Processing summary """ total = len(files) for i, file in enumerate(files): # Report progress (0-100%) await ctx.report_progress(i * 100 // total) # Process the file... ctx.info(f"Processing {file}") return f"Processed {total} files" ``` ### Lifespan Management For servers that need initialization and cleanup: ```python from contextlib import asynccontextmanager from typing import AsyncIterator @asynccontextmanager async def lifespan(server: FastMCP) -> AsyncIterator[dict]: """Manage server lifecycle.""" # Setup (runs on startup) db = await Database.connect() try: yield {"db": db} # Pass context to handlers finally: # Cleanup (runs on shutdown) await db.disconnect() # Create server with lifespan mcp = FastMCP("DatabaseServer", lifespan=lifespan) @mcp.tool() async def query_db(sql: str, ctx: Context) -> list: """Run a database query.""" db = ctx.request_context.lifespan_context["db"] return await db.execute(sql) ``` ## Testing MCP Servers ### Using the MCP Inspector The MCP Inspector is a tool for testing MCP servers: ```bash # Install the inspector npm install -g @modelcontextprotocol/inspector # Run your server with the inspector npx @modelcontextprotocol/inspector python server.py ``` This opens a web interface where you can: - See available tools, resources, and prompts - Test tools with different parameters - View tool execution results - Explore resource content ### Manual Testing You can also test your server programmatically: ```python import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def test_server(): # Connect to the server server_params = StdioServerParameters( command="python", args=["server.py"] ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # Initialize the connection await session.initialize() # List tools tools = await session.list_tools() print(f"Available tools: {[tool.name for tool in tools.tools]}") # Call a tool result = await session.call_tool("hello", {"name": "MCP"}) print(f"Tool result: {result.content[0].text}") if __name__ == "__main__": asyncio.run(test_server()) ``` ## Debugging MCP Servers ### Logging Use logging to debug your server: ```python import logging # Configure logging logging.basicConfig( level=logging.DEBUG, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) # Access the MCP logger logger = logging.getLogger("mcp") ``` ### Common Issues 1. **Schema Generation**: - Ensure type hints are accurate - Provide docstrings for tools - Check parameter names and types 2. **Async/Sync Mismatch**: - Use `async def` for tools that use async operations - Don't mix async and sync code without proper handling 3. **Transport Issues**: - Check that stdio is not mixed with print statements - Ensure ports are available for SSE transport - Verify network settings for remote connections ## Deployment Options ### Local Deployment For local use with Claude Desktop: 1. Edit the Claude Desktop config file: ```json { "mcpServers": { "my-server": { "command": "python", "args": ["/path/to/server.py"] } } } ``` 2. Restart Claude Desktop ### Web Deployment For web deployment with SSE transport: 1. Set up a web server (e.g., nginx) to proxy requests 2. Use a process manager (e.g., systemd, supervisor) to keep the server running 3. Configure the server to use SSE transport with appropriate host/port Example systemd service: ```ini [Unit] Description=MCP Server After=network.target [Service] User=mcp WorkingDirectory=/path/to/server ExecStart=/path/to/venv/bin/python server.py --transport sse --host 127.0.0.1 --port 5000 Restart=on-failure [Install] WantedBy=multi-user.target ``` ## Security Considerations When building MCP servers, consider these security aspects: 1. **Input Validation**: - Validate all parameters - Sanitize file paths and system commands - Use allowlists for sensitive operations 2. **Resource Access**: - Limit access to specific directories - Avoid exposing sensitive information - Use proper permissions for files 3. **Error Handling**: - Don't expose internal errors to clients - Log security-relevant errors - Implement proper error recovery 4. **Authentication**: - Implement authentication for sensitive operations - Use secure tokens or credentials - Verify client identity when needed ## Example: Web Scraping Server Let's build a complete web scraping server that fetches and returns content from URLs: ```python # server.py import httpx from mcp.server.fastmcp import FastMCP # Create the server mcp = FastMCP("WebScraper") @mcp.tool() async def web_scrape(url: str) -> str: """ Fetch content from a URL and return it. Args: url: The URL to scrape Returns: The page content """ # Ensure URL has a scheme if not url.startswith(('http://', 'https://')): url = 'https://' + url # Fetch the content try: async with httpx.AsyncClient() as client: response = await client.get(url, follow_redirects=True) response.raise_for_status() return response.text except httpx.HTTPStatusError as e: return f"Error: HTTP status error - {e.response.status_code}" except httpx.RequestError as e: return f"Error: Request failed - {str(e)}" except Exception as e: return f"Error: Unexpected error occurred - {str(e)}" if __name__ == "__main__": mcp.run() ``` ## Conclusion Building MCP servers with Python is a powerful way to extend LLM capabilities. By following the patterns and practices in this guide, you can create robust, maintainable MCP servers that integrate with Claude and other LLMs. In the next document, we'll explore how to connect to MCP servers from different clients.