### Install MCP Client Source: https://github.com/gwbischof/python-mcp/blob/main/examples/clients/simple-auth-client/README.md Installs the MCP Python SDK client, including its dependencies, using uv sync. This command should be run within the client's example directory. ```bash cd examples/clients/simple-auth-client uv sync --reinstall ``` -------------------------------- ### Start MCP Server Source: https://github.com/gwbischof/python-mcp/blob/main/examples/clients/simple-auth-client/README.md Starts an MCP server with OAuth support using uv run. This command requires navigating to the mcp-simple-auth directory and specifies the transport protocol and port. ```bash # Example with mcp-simple-auth cd path/to/mcp-simple-auth uv run mcp-simple-auth --transport streamable-http --port 3001 ``` -------------------------------- ### Start MCP Simple Tool Server (Bash) Source: https://github.com/gwbischof/python-mcp/blob/main/examples/servers/simple-tool/README.md Instructions to start the MCP simple tool server using the 'uv' command. It demonstrates starting with the default stdio transport and with SSE transport on a custom port. ```bash # Using stdio transport (default) uv run mcp-simple-tool # Using SSE transport on custom port uv run mcp-simple-tool --transport sse --port 8000 ``` -------------------------------- ### Run MCP Simple StreamableHttp Server (Bash) Source: https://github.com/gwbischof/python-mcp/blob/main/examples/servers/simple-streamablehttp/README.md Demonstrates how to start the StreamableHttp MCP server using the 'uv' command-line tool. Supports custom ports, logging levels, and JSON responses. ```bash uv run mcp-simple-streamablehttp --port 3000 uv run mcp-simple-streamablehttp --log-level DEBUG uv run mcp-simple-streamablehttp --json-response ``` -------------------------------- ### Start MCP Simple Resource Server (Bash) Source: https://github.com/gwbischof/python-mcp/blob/main/examples/servers/simple-resource/README.md Instructions for starting the MCP Simple Resource server using the `uv` command. Supports default stdio transport and custom SSE transport with a specified port. ```bash # Using stdio transport (default) uv run mcp-simple-resource # Using SSE transport on custom port uv run mcp-simple-resource --transport sse --port 8000 ``` -------------------------------- ### Install and Develop MCP Server Source: https://github.com/gwbischof/python-mcp/blob/main/README.md Commands to install a Python MCP server for use with Claude Desktop or to run it in development mode using the MCP Inspector. ```bash mcp install server.py ``` ```bash mcp dev server.py ``` -------------------------------- ### MCP CLI: Development and Installation Commands Source: https://github.com/gwbischof/python-mcp/blob/main/README.md These commands are used to run MCP servers in development mode, install them into Claude Desktop with custom configurations, and manage dependencies or environment variables. ```bash mcp dev server.py # Add dependencies mcp dev server.py --with pandas --with numpy # Mount local code mcp dev server.py --with-editable . mcp install server.py # Custom name mcp install server.py --name "My Analytics Server" # Environment variables mcp install server.py -v API_KEY=abc123 -v DB_URL=postgres://... mcp install server.py -f .env ``` -------------------------------- ### Install MCP Python SDK Source: https://github.com/gwbischof/python-mcp/blob/main/README.md Installs the MCP Python SDK with CLI tools using uv or pip. It's recommended to use uv for project management. ```bash uv init mcp-server-demo cd mcp-server-demo uv add "mcp[cli]" ``` ```bash pip install "mcp[cli]" ``` -------------------------------- ### Install Python Dependencies Source: https://github.com/gwbischof/python-mcp/blob/main/examples/clients/simple-chatbot/README.MD Installs all required Python packages for the MCP chatbot project using pip and a requirements.txt file. This is a standard first step for setting up Python projects. ```bash pip install -r requirements.txt ``` -------------------------------- ### Run MCP Chatbot Client Source: https://github.com/gwbischof/python-mcp/blob/main/examples/clients/simple-chatbot/README.MD Executes the main Python script to start the MCP chatbot. Once running, the chatbot will automatically discover and utilize configured tools. ```bash python main.py ``` -------------------------------- ### Interact with MCP Simple Tool Server (Python) Source: https://github.com/gwbischof/python-mcp/blob/main/examples/servers/simple-tool/README.md An example Python script demonstrating how to use the MCP client to connect to the simple tool server via stdio transport. It shows how to initialize the client session, list available tools, and call the 'fetch' tool with a URL. ```python import asyncio from mcp.client.session import ClientSession from mcp.client.stdio import StdioServerParameters, stdio_client async def main(): async with stdio_client( StdioServerParameters(command="uv", args=["run", "mcp-simple-tool"]) ) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # List available tools tools = await session.list_tools() print(tools) # Call the fetch tool result = await session.call_tool("fetch", {"url": "https://example.com"}) print(result) asyncio.run(main()) ``` -------------------------------- ### Create a Simple MCP Server (Python) Source: https://github.com/gwbischof/python-mcp/blob/main/README.md Demonstrates the creation of a basic MCP server using FastMCP. This server exposes an 'add' tool and a dynamic 'greeting' resource. It can be installed in Claude Desktop or tested with the MCP Inspector. ```python # server.py from mcp.server.fastmcp import FastMCP # Create an MCP server mcp = FastMCP("Demo") # Add an addition tool @mcp.tool() def add(a: int, b: int) -> int: """Add two numbers""" return a + b # Add a dynamic greeting resource @mcp.resource("greeting://{name}") def get_greeting(name: str) -> str: """Get a personalized greeting""" return f"Hello, {name}!" ``` -------------------------------- ### Run MCP Simple Prompt Server Source: https://github.com/gwbischof/python-mcp/blob/main/examples/servers/simple-prompt/README.md Commands to start the MCP Simple Prompt server using stdio or SSE transport. Stdio is the default, while SSE requires specifying a port. ```bash # Using stdio transport (default) uv run mcp-simple-prompt # Using SSE transport on custom port uv run mcp-simple-prompt --transport sse --port 8000 ``` -------------------------------- ### Install Dependencies with uv Source: https://github.com/gwbischof/python-mcp/blob/main/CONTRIBUTING.md Installs project dependencies using the 'uv' package manager. It ensures that all extras and development dependencies are included and uses a frozen lock file for reproducible builds. This command is essential for setting up the development environment. ```bash uv sync --frozen --all-extras --dev ``` -------------------------------- ### Run MCP Client Source: https://github.com/gwbischof/python-mcp/blob/main/examples/clients/simple-auth-client/README.md Runs the simple MCP authentication client using uv run. It can be executed with the default server URL or a custom URL specified via the MCP_SERVER_URL environment variable. ```bash uv run mcp-simple-auth-client # Or with custom server URL MCP_SERVER_URL=http://localhost:3001 uv run mcp-simple-auth-client ``` -------------------------------- ### Python: Configuring Mount Paths for Multiple MCP Servers Source: https://github.com/gwbischof/python-mcp/blob/main/README.md This example illustrates two methods for configuring the mount paths of multiple MCP servers within an ASGI application: setting them in settings or passing them directly to the sse_app method. ```python from starlette.applications import Starlette from starlette.routing import Mount from mcp.server.fastmcp import FastMCP # Create multiple MCP servers github_mcp = FastMCP("GitHub API") browser_mcp = FastMCP("Browser") curl_mcp = FastMCP("Curl") search_mcp = FastMCP("Search") # Method 1: Configure mount paths via settings (recommended for persistent configuration) github_mcp.settings.mount_path = "/github" browser_mcp.settings.mount_path = "/browser" # Method 2: Pass mount path directly to sse_app (preferred for ad-hoc mounting) # This approach doesn't modify the server's settings permanently ``` -------------------------------- ### Package Management with uv Source: https://github.com/gwbischof/python-mcp/blob/main/CLAUDE.md Instructions for managing Python packages using the `uv` tool, including installation, running tools, and upgrading packages. It explicitly forbids the use of `pip` and certain `uv` syntaxes. ```bash uv add package ``` ```bash uv run tool ``` ```bash uv add --dev package --upgrade-package package ``` -------------------------------- ### Interact with MCP Server using Python Client (Python) Source: https://github.com/gwbischof/python-mcp/blob/main/examples/servers/simple-resource/README.md Example Python code demonstrating how to use the MCP client to interact with the Simple Resource server. It initializes a client session, lists available resources, and reads a specific resource ('greeting.txt') via stdio transport. ```python import asyncio from mcp.types import AnyUrl from mcp.client.session import ClientSession from mcp.client.stdio import StdioServerParameters, stdio_client async def main(): async with stdio_client( StdioServerParameters(command="uv", args=["run", "mcp-simple-resource"]) ) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # List available resources resources = await session.list_resources() print(resources) # Get a specific resource resource = await session.read_resource(AnyUrl("file:///greeting.txt")) print(resource) asyncio.run(main()) ``` -------------------------------- ### Echo Server using FastMCP Source: https://github.com/gwbischof/python-mcp/blob/main/README.md A basic server example using `FastMCP` to demonstrate resources, tools, and prompts. It defines an echo resource, an echo tool, and an echo prompt, all of which return an echoed version of the input message. ```python from mcp.server.fastmcp import FastMCP mcp = FastMCP("Echo") @mcp.resource("echo://{message}") def echo_resource(message: str) -> str: """Echo a message as a resource""" return f"Resource echo: {message}" @mcp.tool() def echo_tool(message: str) -> str: """Echo a message as a tool""" return f"Tool echo: {message}" @mcp.prompt() def echo_prompt(message: str) -> str: """Create an echo prompt""" return f"Please process this message: {message}" ``` -------------------------------- ### FastMCP Prompt Definition Source: https://github.com/gwbischof/python-mcp/blob/main/README.md Demonstrates defining reusable prompts in FastMCP for effective LLM interaction. Includes examples for code review and debugging assistance, utilizing different message types. ```python from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp.prompts import base mcp = FastMCP("My App") @mcp.prompt() def review_code(code: str) -> str: return f"Please review this code:\n\n{code}" @mcp.prompt() def debug_error(error: str) -> list[base.Message]: return [ base.UserMessage("I'm seeing this error:"), base.UserMessage(error), base.AssistantMessage("I'll help debug that. What have you tried so far?"), ] ``` -------------------------------- ### Initialize Client Sessions with Stdio and HTTP Transports Source: https://context7.com/gwbischof/python-mcp/llms.txt This snippet shows how to initialize client sessions to connect to MCP servers using different transport mechanisms. It covers connecting via standard input/output (stdio) for local servers, including setting environment variables and executing commands. It also demonstrates connecting via streamable HTTP for web servers. Both examples show how to initialize the session, list tools and resources, call tools, and read resources. ```python from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from mcp.client.streamable_http import streamablehttp_client import asyncio # Stdio client (for local servers) async def connect_stdio(): server_params = StdioServerParameters( command="python", args=["server.py"], env={"API_KEY": "secret"} ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # Initialize connection result = await session.initialize() print(f"Connected to: {result.serverInfo.name}") # List available tools tools = await session.list_tools() for tool in tools.tools: print(f"Tool: {tool.name} - {tool.description}") # Call a tool tool_result = await session.call_tool( "add", arguments={"a": 5, "b": 3} ) print(f"Result: {tool_result}") # List resources resources = await session.list_resources() for resource in resources.resources: print(f"Resource: {resource.uri}") # Read a resource content = await session.read_resource("config://app") print(f"Config: {content}") # Streamable HTTP client (for web servers) async def connect_http(): async with streamablehttp_client("https://example.com/mcp") as ( read, write, _ ): async with ClientSession(read, write) as session: await session.initialize() # Use session result = await session.call_tool( "echo", arguments={"message": "hello"} ) print(result) asyncio.run(connect_stdio()) ``` -------------------------------- ### Implement Low-Level MCP Server (Python) Source: https://context7.com/gwbischof/python-mcp/llms.txt Provides a foundation for building MCP servers with complete control over protocol implementation and lifecycle management. It demonstrates using `asynccontextmanager` for server lifespan management and registering tool handlers using the `@server.call_tool()` decorator. The example includes a basic tool handler for querying a database. ```Python from contextlib import asynccontextmanager from collections.abc import AsyncIterator import mcp.server.stdio import mcp.types as types from mcp.server.lowlevel import Server, NotificationOptions from mcp.server.models import InitializationOptions import asyncio # Lifespan management @asynccontextmanager async def server_lifespan(server: Server) -> AsyncIterator[dict]: """Initialize resources on startup""" db = await connect_database() # Assuming connect_database is defined elsewhere try: yield {"db": db} finally: await db.disconnect() # Create server with lifespan server = Server("example-server", lifespan=server_lifespan) # Register tool handler @server.call_tool() async def handle_call_tool(name: str, arguments: dict) -> list: """Handle tool calls""" ctx = server.get_context() db = ctx.lifespan_context["db"] if name == "query_database": results = await db.execute(arguments["query"]) return [ types.TextContent( type="text", text=f"Query results: {results}" ) ] raise ValueError(f"Unknown tool: {name}") # To run the server, you would typically add: # async def main(): # await server.run_forever() # asyncio.run(main()) ``` -------------------------------- ### SQLite Explorer using FastMCP Source: https://github.com/gwbischof/python-mcp/blob/main/README.md An example of integrating SQLite with `FastMCP`. It provides the database schema as a resource and allows executing SQL queries safely through a tool. It connects to a `database.db` file. ```python import sqlite3 from mcp.server.fastmcp import FastMCP mcp = FastMCP("SQLite Explorer") @mcp.resource("schema://main") def get_schema() -> str: """Provide the database schema as a resource""" conn = sqlite3.connect("database.db") schema = conn.execute("SELECT sql FROM sqlite_master WHERE type='table'").fetchall() return "\n".join(sql[0] for sql in schema if sql[0]) @mcp.tool() def query_data(sql: str) -> str: """Execute SQL queries safely""" conn = sqlite3.connect("database.db") try: result = conn.execute(sql).fetchall() return "\n".join(str(row) for row in result) except Exception as e: return f"Error: {str(e)}" ``` -------------------------------- ### FastMCP Tool Definition Source: https://github.com/gwbischof/python-mcp/blob/main/README.md Illustrates how to define tools in FastMCP, enabling LLMs to perform actions and have side effects. Examples include a BMI calculator and an asynchronous weather fetcher using httpx. ```python import httpx from mcp.server.fastmcp import FastMCP mcp = FastMCP("My App") @mcp.tool() def calculate_bmi(weight_kg: float, height_m: float) -> float: """Calculate BMI given weight in kg and height in meters""" return weight_kg / (height_m**2) @mcp.tool() async def fetch_weather(city: str) -> str: """Fetch current weather for a city""" async with httpx.AsyncClient() as client: response = await client.get(f"https://api.weather.com/{city}") return response.text ``` -------------------------------- ### FastAPI: Mounting Multiple MCP Servers Source: https://github.com/gwbischof/python-mcp/blob/main/README.md This example demonstrates how to mount multiple independent MCP servers (EchoServer and MathServer) as separate endpoints within a single FastAPI application using streamable HTTP. ```python # echo.py from mcp.server.fastmcp import FastMCP mcp = FastMCP(name="EchoServer", stateless_http=True) @mcp.tool(description="A simple echo tool") def echo(message: str) -> str: return f"Echo: {message}" ``` ```python # math.py from mcp.server.fastmcp import FastMCP mcp = FastMCP(name="MathServer", stateless_http=True) @mcp.tool(description="A simple add tool") def add_two(n: int) -> int: return n + 2 ``` ```python # main.py import contextlib from fastapi import FastAPI from mcp.echo import echo from mcp.math import math # Create a combined lifespan to manage both session managers @contextlib.asynccontextmanager async def lifespan(app: FastAPI): async with contextlib.AsyncExitStack() as stack: await stack.enter_async_context(echo.mcp.session_manager.run()) await stack.enter_async_context(math.mcp.session_manager.run()) yield app = FastAPI(lifespan=lifespan) app.mount("/echo", echo.mcp.streamable_http_app()) app.mount("/math", math.mcp.streamable_http_app()) ``` -------------------------------- ### FastMCP Context for Tools and Resources Source: https://github.com/gwbischof/python-mcp/blob/main/README.md Details the `Context` object in FastMCP, providing tools and resources access to MCP capabilities like logging, progress reporting, and resource reading. The example demonstrates processing multiple files with progress tracking. ```python from mcp.server.fastmcp import FastMCP, Context mcp = FastMCP("My App") @mcp.tool() async def long_task(files: list[str], ctx: Context) -> str: """Process multiple files with progress tracking""" for i, file in enumerate(files): ctx.info(f"Processing {file}") await ctx.report_progress(i, len(files)) data, mime_type = await ctx.read_resource(f"file://{file}") return "Processing complete" ``` -------------------------------- ### Write MCP Client with Streamable HTTP Transport in Python Source: https://github.com/gwbischof/python-mcp/blob/main/README.md This Python snippet shows how to establish a connection to an MCP server using the streamable HTTP transport. It outlines the process of creating a client session using the provided streams and then initializing the connection to the server, followed by an example of calling a tool. ```python from mcp.client.streamable_http import streamablehttp_client from mcp import ClientSession async def main(): # Connect to a streamable HTTP server async with streamablehttp_client("example/mcp") as ( read_stream, write_stream, _, ): # Create a session using the client streams async with ClientSession(read_stream, write_stream) as session: # Initialize the connection await session.initialize() # Call a tool tool_result = await session.call_tool("echo", {"message": "hello"}) ``` -------------------------------- ### Running MCP Server with Streamable HTTP Transport Source: https://github.com/gwbischof/python-mcp/blob/main/examples/servers/simple-auth/README.md This command starts the MCP server using the Streamable HTTP transport protocol. This option provides the '/mcp' endpoint for communication, offering an alternative to SSE. Ensure this transport is compatible with your client-side implementation. ```bash uv run mcp-simple-auth --transport streamable-http ``` -------------------------------- ### Running MCP Server with SSE Transport Source: https://github.com/gwbischof/python-mcp/blob/main/examples/servers/simple-auth/README.md This command explicitly starts the MCP server using the Server-Sent Events (SSE) transport protocol. SSE is the default transport if none is specified. This configuration makes the '/sse' endpoint available for communication. ```bash uv run mcp-simple-auth --transport sse ``` -------------------------------- ### Mounting Multiple FastMCP Servers in FastAPI Source: https://context7.com/gwbischof/python-mcp/llms.txt Provides examples of mounting multiple independent FastMCP servers within a single FastAPI application to create a microservices architecture. This allows for modular development and deployment. Requires `mcp.server.fastmcp.FastMCP`. ```python # echo_server.py from mcp.server.fastmcp import FastMCP echo_mcp = FastMCP(name="EchoServer", stateless_http=True) @echo_mcp.tool(description="Echo a message") def echo(message: str) -> str: return f"Echo: {message}" # math_server.py from mcp.server.fastmcp import FastMCP math_mcp = FastMCP(name="MathServer", stateless_http=True) @math_mcp.tool(description="Add two numbers") def add(a: int, b: int) -> int: return a + b ``` -------------------------------- ### FastMCP OAuth 2.0 Server Configuration Source: https://github.com/gwbischof/python-mcp/blob/main/README.md Provides an example of configuring an OAuth 2.0 server within FastMCP for authentication. This involves providing an implementation of `OAuthServerProvider` and setting up authentication settings like issuer URL and client registration options. ```python mcp = FastMCP("My App", auth_server_provider=MyOAuthServerProvider(), auth=AuthSettings( issuer_url="https://myapp.com", revocation_options=RevocationOptions( enabled=True, ), client_registration_options=ClientRegistrationOptions( enabled=True, valid_scopes=["myscope", "myotherscope"], default_scopes=["myscope"], ), required_scopes=["myscope"], ), ) ``` -------------------------------- ### Write MCP Client with Stdio Transport in Python Source: https://github.com/gwbischof/python-mcp/blob/main/README.md This Python code demonstrates how to create an MCP client that connects to an MCP server using the stdio transport. It includes setting up server parameters, defining an optional sampling callback, and performing various client operations like initializing the connection, listing prompts, getting prompts, and calling tools. ```python from mcp import ClientSession, StdioServerParameters, types from mcp.client.stdio import stdio_client # Create server parameters for stdio connection server_params = StdioServerParameters( command="python", # Executable args=["example_server.py"], # Optional command line arguments env=None, # Optional environment variables ) # Optional: create a sampling callback async def handle_sampling_message( message: types.CreateMessageRequestParams, ) -> types.CreateMessageResult: return types.CreateMessageResult( role="assistant", content=types.TextContent( type="text", text="Hello, world! from model", ), model="gpt-3.5-turbo", stopReason="endTurn", ) async def run(): async with stdio_client(server_params) as (read, write): async with ClientSession( read, write, sampling_callback=handle_sampling_message ) as session: # Initialize the connection await session.initialize() # List available prompts prompts = await session.list_prompts() # Get a prompt prompt = await session.get_prompt( "example-prompt", arguments={"arg1": "value"} ) # List available resources resources = await session.list_resources() # List available tools tools = await session.list_tools() # Read a resource content, mime_type = await session.read_resource("file://some/path") # Call a tool result = await session.call_tool("tool-name", arguments={"arg1": "value"}) if __name__ == "__main__": import asyncio asyncio.run(run()) ``` -------------------------------- ### Running the MCP Server Source: https://github.com/gwbischof/python-mcp/blob/main/examples/servers/simple-auth/README.md This command initiates the MCP server using the 'uv' command-line tool. It assumes that the required environment variables for GitHub OAuth have been set beforehand. The server will then be accessible at http://localhost:8000. ```bash # Set environment variables first (see above) # Run the server uv run mcp-simple-auth ``` -------------------------------- ### Running Pytest with AnyIO and Environment Variable Source: https://github.com/gwbischof/python-mcp/blob/main/CLAUDE.md Guidance on how to run Pytest, particularly when dealing with AnyIO for async testing. It includes a specific command to prepend an environment variable to resolve potential plugin autoloading issues. ```bash uv run --frozen pytest ``` ```bash PYTEST_DISABLE_PLUGIN_AUTOLOAD="" uv run --frozen pytest ``` -------------------------------- ### Create MCP Server Instance in Python Source: https://github.com/gwbischof/python-mcp/blob/main/README.md This snippet demonstrates how to create a new MCP server instance, define handlers for listing prompts, and retrieving prompt details. It sets up the server to respond to specific requests related to prompts and includes a function to run the server with stdio transport. ```python from mcp import types # Create a server instance server = Server("example-server") @server.list_prompts() async def handle_list_prompts() -> list[types.Prompt]: return [ types.Prompt( name="example-prompt", description="An example prompt template", arguments=[ types.PromptArgument( name="arg1", description="Example argument", required=True ) ], ) ] @server.get_prompt() async def handle_get_prompt( name: str, arguments: dict[str, str] | None ) -> types.GetPromptResult: if name != "example-prompt": raise ValueError(f"Unknown prompt: {name}") return types.GetPromptResult( description="Example prompt", messages=[ types.PromptMessage( role="user", content=types.TextContent(type="text", text="Example prompt text"), ) ], ) async def run(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_name="example", server_version="0.1.0", capabilities=server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={}, ), ), ) if __name__ == "__main__": import asyncio asyncio.run(run()) ``` -------------------------------- ### Initialize FastMCP Server Source: https://context7.com/gwbischof/python-mcp/llms.txt Demonstrates how to initialize a FastMCP server with options for dependencies, lifecycle management, and authentication. Requires the 'mcp' library. ```python from mcp.server.fastmcp import FastMCP, Context from contextlib import asynccontextmanager from collections.abc import AsyncIterator from dataclasses import dataclass # Simple server mcp = FastMCP("My App") # Server with dependencies mcp = FastMCP("My App", dependencies=["pandas", "numpy"]) # Server with lifespan management @dataclass class AppContext: db: object # Your database connection @asynccontextmanager async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]: """Initialize resources on startup, cleanup on shutdown""" db = await connect_to_database() try: yield AppContext(db=db) finally: await db.disconnect() mcp = FastMCP("My App", lifespan=app_lifespan) # Server with authentication from mcp.server.auth.settings import AuthSettings, ClientRegistrationOptions mcp = FastMCP( "My App", auth_server_provider=MyOAuthServerProvider(), auth=AuthSettings( issuer_url="https://myapp.com", client_registration_options=ClientRegistrationOptions( enabled=True, valid_scopes=["read", "write"], default_scopes=["read"], ), required_scopes=["read"], ), ) ``` -------------------------------- ### FastMCP Server Initialization and Lifespan Management Source: https://github.com/gwbischof/python-mcp/blob/main/README.md Demonstrates initializing the FastMCP server with a name, optional dependencies, and lifespan support for startup and shutdown operations. It includes type-safe context management for resources like databases. ```python # Add lifespan support for startup/shutdown with strong typing from contextlib import asynccontextmanager from collections.abc import AsyncIterator from dataclasses import dataclass from fake_database import Database # Replace with your actual DB type from mcp.server.fastmcp import Context, FastMCP # Create a named server mcp = FastMCP("My App") # Specify dependencies for deployment and development mcp = FastMCP("My App", dependencies=["pandas", "numpy"]) @dataclass class AppContext: db: Database @asynccontextmanager async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]: """Manage application lifecycle with type-safe context""" # Initialize on startup db = await Database.connect() try: yield AppContext(db=db) finally: # Cleanup on shutdown await db.disconnect() # Pass lifespan to server mcp = FastMCP("My App", lifespan=app_lifespan) # Access type-safe lifespan context in tools @mcp.tool() def query_db(ctx: Context) -> str: """Tool that uses initialized resources""" db = ctx.request_context.lifespan_context.db return db.query() ``` -------------------------------- ### FastMCP Image Handling Source: https://github.com/gwbischof/python-mcp/blob/main/README.md Explains the `Image` class in FastMCP for automatic image data handling. The example shows creating a thumbnail from an image file using Pillow and returning it as a FastMCP `Image` object. ```python from mcp.server.fastmcp import FastMCP, Image from PIL import Image as PILImage mcp = FastMCP("My App") @mcp.tool() def create_thumbnail(image_path: str) -> Image: """Create a thumbnail from an image""" img = PILImage.open(image_path) img.thumbnail((100, 100)) return Image(data=img.tobytes(), format="png") ``` -------------------------------- ### Low-Level Server with Lifespan API Source: https://github.com/gwbischof/python-mcp/blob/main/README.md Demonstrates advanced usage of the low-level `Server` implementation in MCP. It shows how to manage server lifecycle (startup and shutdown) using an async context manager for resource initialization and cleanup, and how to access these resources within handlers. ```python from contextlib import asynccontextmanager from collections.abc import AsyncIterator from fake_database import Database # Replace with your actual DB type from mcp.server import Server @asynccontextmanager async def server_lifespan(server: Server) -> AsyncIterator[dict]: """Manage server startup and shutdown lifecycle.""" # Initialize resources on startup db = await Database.connect() try: yield {"db": db} finally: # Clean up on shutdown await db.disconnect() # Pass lifespan to server server = Server("example-server", lifespan=server_lifespan) # Access lifespan context in handlers @server.call_tool() async def query_db(name: str, arguments: dict) -> list: ctx = server.get_context() db = ctx.lifespan_context["db"] return await db.query(arguments["query"]) ``` ```python import mcp.server.stdio import mcp.types as types from mcp.server.lowlevel import NotificationOptions, Server from mcp.server.models import InitializationOptions ``` -------------------------------- ### FastMCP Resource Definition Source: https://github.com/gwbischof/python-mcp/blob/main/README.md Shows how to define resources in FastMCP, which are used to expose data similar to GET endpoints in REST APIs. Resources are typically static or dynamic data providers without side effects. ```python from mcp.server.fastmcp import FastMCP mcp = FastMCP("My App") @mcp.resource("config://app") def get_config() -> str: """Static configuration data""" return "App configuration here" @mcp.resource("users://{user_id}/profile") def get_user_profile(user_id: str) -> str: """Dynamic user data""" return f"Profile data for user {user_id}" ``` -------------------------------- ### Create Starlette App with Mounted Servers Source: https://github.com/gwbischof/python-mcp/blob/main/README.md Demonstrates how to create a Starlette application and mount multiple other applications (servers) under different URL paths. This utilizes the `Mount` functionality from Starlette for routing. ```python from starlette.applications import Starlette from starlette.routing import Mount # Assuming github_mcp, browser_mcp, curl_mcp, and search_mcp are defined elsewhere # from . import github_mcp # from . import browser_mcp # from . import curl_mcp # from . import search_mcp app = Starlette( routes=[ # Using settings-based configuration Mount("/github", app=github_mcp.sse_app()), Mount("/browser", app=browser_mcp.sse_app()), # Using direct mount path parameter Mount("/curl", app=curl_mcp.sse_app("/curl")), Mount("/search", app=search_mcp.sse_app("/search")), ] ) # Method 3: For direct execution, you can also pass the mount path to run() if __name__ == "__main__": search_mcp.run(transport="sse", mount_path="/search") ``` -------------------------------- ### MCP Python: Direct Execution with FastMCP Source: https://github.com/gwbischof/python-mcp/blob/main/README.md This Python snippet demonstrates how to instantiate and run a simple MCP server using the FastMCP class. It can be executed directly using Python or the 'mcp run' command. ```python from mcp.server.fastmcp import FastMCP mcp = FastMCP("My App") if __name__ == "__main__": mcp.run() ``` -------------------------------- ### Register Resources for Data Access Source: https://context7.com/gwbischof/python-mcp/llms.txt Illustrates how to register data sources as resources that LLMs can access. Supports static configuration data and dynamic resources with parameters, requiring 'mcp'. ```python from mcp.server.fastmcp import FastMCP import json import sqlite3 mcp = FastMCP("Data Server") # Static resource @mcp.resource("config://app") def get_config() -> str: """Provide application configuration""" return json.dumps({ "version": "1.0.0", "features": ["tool_calling", "resources"] }) # Template resource with dynamic parameters @mcp.resource("users://{user_id}/profile") def get_user_profile(user_id: str) -> str: """Get user profile data""" conn = sqlite3.connect("database.db") cursor = conn.execute( "SELECT name, email, created_at FROM users WHERE id = ?", (user_id,) ) row = cursor.fetchone() if row: return json.dumps({ "name": row[0], "email": row[1], "created_at": row[2] }) return json.dumps({"error": "User not found"}) ``` -------------------------------- ### Interact with MCP Simple Prompt Server using Python Client Source: https://github.com/gwbischof/python-mcp/blob/main/examples/servers/simple-prompt/README.md Python code to interact with the MCP Simple Prompt server using the MCP client library. It shows how to establish a connection via stdio, list available prompts, and retrieve a specific prompt with context and topic arguments. ```python import asyncio from mcp.client.session import ClientSession from mcp.client.stdio import StdioServerParameters, stdio_client async def main(): async with stdio_client( StdioServerParameters(command="uv", args=["run", "mcp-simple-prompt"]) ) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # List available prompts prompts = await session.list_prompts() print(prompts) # Get the prompt with arguments prompt = await session.get_prompt( "simple", { "context": "User is a software developer", "topic": "Python async programming", }, ) print(prompt) asyncio.run(main()) ``` -------------------------------- ### ASGI: Mounting MCP SSE Server with Starlette Source: https://github.com/gwbischof/python-mcp/blob/main/README.md This Python code shows how to mount an MCP SSE server into an existing Starlette ASGI application. It demonstrates mounting the SSE app directly or dynamically as a host. ```python from starlette.applications import Starlette from starlette.routing import Mount, Host from mcp.server.fastmcp import FastMCP mcp = FastMCP("My App") # Mount the SSE server to the existing ASGI server app = Starlette( routes=[ Mount('/', app=mcp.sse_app()), ] ) # or dynamically mount as host app.router.routes.append(Host('mcp.acme.corp', app=mcp.sse_app())) ``` -------------------------------- ### Configure MCP Servers with JSON Source: https://github.com/gwbischof/python-mcp/blob/main/examples/clients/simple-chatbot/README.MD Defines the configuration for MCP servers, similar to Claude Desktop's structure. This JSON file specifies servers, their commands, arguments, and optional environment variables for integration. ```json { "mcpServers": { "sqlite": { "command": "uvx", "args": ["mcp-server-sqlite", "--db-path", "./test.db"] }, "puppeteer": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-puppeteer"] } } } ``` ```json { "mcpServers": { "server_name": { "command": "uvx", "args": ["mcp-server-name", "--additional-args"], "env": { "API_KEY": "your_api_key_here" } } } } ``` -------------------------------- ### Setting Environment Variables for GitHub OAuth Source: https://github.com/gwbischof/python-mcp/blob/main/examples/servers/simple-auth/README.md These commands set the necessary environment variables for GitHub OAuth client ID and client secret. These variables are mandatory for the server to authenticate with GitHub. Ensure you replace the placeholder values with your actual credentials obtained from your GitHub OAuth App. ```bash export MCP_GITHUB_GITHUB_CLIENT_ID="your_client_id_here" export MCP_GITHUB_GITHUB_CLIENT_SECRET="your_client_secret_here" ``` -------------------------------- ### Run MCP Development Tools Source: https://github.com/gwbischof/python-mcp/blob/main/README.md Executes the MCP command-line interface tools within a uv-managed environment. ```bash uv run mcp ``` -------------------------------- ### Type Checking with Pyright Source: https://github.com/gwbischof/python-mcp/blob/main/CLAUDE.md Instructions for running Pyright for type checking Python code. It highlights requirements such as explicit None checks for Optional types and type narrowing for strings. ```bash uv run --frozen pyright ``` -------------------------------- ### Async File Reading and Downloading with MCP Source: https://context7.com/gwbischof/python-mcp/llms.txt Demonstrates how to create asynchronous resources in MCP for reading plain text files and downloading binary files. It utilizes `aiofiles` for non-blocking file operations and specifies MIME types for different content. Requires the `aiofiles` library. ```python import aiofiles from mcp.server.fastmcp import FastMCP mcp = FastMCP("FileServer") @mcp.resource("files://{path}/content", mime_type="text/plain") async def read_file(path: str) -> str: """Read file contents""" async with aiofiles.open(path, 'r') as f: content = await f.read() return content @mcp.resource("files://{path}/download", mime_type="application/octet-stream") async def download_file(path: str) -> bytes: """Download binary file""" async with aiofiles.open(path, 'rb') as f: return await f.read() ``` -------------------------------- ### Running FastMCP Servers with Different Transports Source: https://context7.com/gwbischof/python-mcp/llms.txt Demonstrates how to run FastMCP servers using various transport protocols, including 'stdio' for local development, 'sse' for web integration, and 'streamable-http' for production. It also shows configurations for stateful, stateless, and stateless servers with JSON responses. Requires `mcp.server.fastmcp.FastMCP`. ```python from mcp.server.fastmcp import FastMCP mcp = FastMCP("My App") @mcp.tool() def hello(name: str) -> str: return f"Hello, {name}!" if __name__ == "__main__": mcp.run(transport="stdio") if __name__ == "__main__": mcp.run(transport="sse", mount_path="/api/mcp") mcp = FastMCP("StatefulServer") mcp.run(transport="streamable-http") mcp = FastMCP("StatelessServer", stateless_http=True) mcp.run(transport="streamable-http") mcp = FastMCP("StatelessJSON", stateless_http=True, json_response=True) mcp.run(transport="streamable-http") ``` -------------------------------- ### Mount MCP Servers into Starlette/FastAPI Applications Source: https://context7.com/gwbischof/python-mcp/llms.txt This section illustrates how to integrate MCP servers with existing Starlette or FastAPI applications. It shows three methods for mounting MCP servers: configuring mount paths via settings, passing the mount path directly to the `sse_app` method, and mounting as host-based routing. This allows for flexible integration of MCP functionalities into existing web frameworks. ```python from starlette.applications import Starlette from starlette.routing import Mount, Host from mcp.server.fastmcp import FastMCP # Create multiple MCP servers github_mcp = FastMCP("GitHub API") browser_mcp = FastMCP("Browser") search_mcp = FastMCP("Search") # Add tools to each server @github_mcp.tool() async def get_repo(name: str) -> str: return f"Repository info for {name}" @browser_mcp.tool() async def navigate(url: str) -> str: return f"Navigating to {url}" # Method 1: Configure mount paths via settings github_mcp.settings.mount_path = "/github" browser_mcp.settings.mount_path = "/browser" # Method 2: Pass mount path directly to sse_app app = Starlette( routes=[ Mount("/github", app=github_mcp.sse_app()), Mount("/browser", app=browser_mcp.sse_app()), Mount("/search", app=search_mcp.sse_app("/search")), ] ) # Method 3: Mount as host-based routing app.router.routes.append( Host('mcp.example.com', app=github_mcp.sse_app()) ) ``` -------------------------------- ### Python OAuth Authentication for MCP Clients Source: https://github.com/gwbischof/python-mcp/blob/main/README.md Demonstrates how to set up OAuth authentication for clients connecting to protected MCP servers using the Python SDK. It includes a custom token storage implementation and shows how to integrate with the streamable HTTP client and client session for authenticated communication. ```python from mcp.client.auth import OAuthClientProvider, TokenStorage from mcp.client.session import ClientSession from mcp.client.streamable_http import streamablehttp_client from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken class CustomTokenStorage(TokenStorage): """Simple in-memory token storage implementation.""" async def get_tokens(self) -> OAuthToken | None: pass async def set_tokens(self, tokens: OAuthToken) -> None: pass async def get_client_info(self) -> OAuthClientInformationFull | None: pass async def set_client_info(self, client_info: OAuthClientInformationFull) -> None: pass async def main(): # Set up OAuth authentication oauth_auth = OAuthClientProvider( server_url="https://api.example.com", client_metadata=OAuthClientMetadata( client_name="My Client", redirect_uris=["http://localhost:3000/callback"], grant_types=["authorization_code", "refresh_token"], response_types=["code"], ), storage=CustomTokenStorage(), redirect_handler=lambda url: print(f"Visit: {url}"), callback_handler=lambda: ("auth_code", None), ) # Use with streamable HTTP client async with streamablehttp_client( "https://api.example.com/mcp", auth=oauth_auth ) as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() # Authenticated session ready ``` -------------------------------- ### Committing with Trailers Source: https://github.com/gwbischof/python-mcp/blob/main/CLAUDE.md Guidelines for adding trailers to Git commit messages for bug fixes, feature additions, or linking to GitHub issues. This helps in tracking contributions and issue resolution. ```bash git commit --trailer "Reported-by:" ``` ```bash git commit --trailer "Github-Issue:#" ``` -------------------------------- ### Register Tools for LLM Invocation Source: https://context7.com/gwbischof/python-mcp/llms.txt Shows how to register Python functions as tools for LLMs to use. Supports synchronous, asynchronous, and context-aware tools, with options for custom names and descriptions. Requires 'mcp' and 'httpx'. ```python from mcp.server.fastmcp import FastMCP, Context import httpx mcp = FastMCP("Calculator") # Simple synchronous tool @mcp.tool() def add(a: int, b: int) -> int: """Add two numbers together""" return a + b # Async tool with external API call @mcp.tool() async def fetch_weather(city: str) -> str: """Fetch current weather for a city""" async with httpx.AsyncClient() as client: response = await client.get( f"https://api.weather.com/v1/current", params={"city": city} ) response.raise_for_status() data = response.json() return f"Temperature in {city}: {data['temp']}°C" # Tool with context for logging and progress @mcp.tool() async def process_files( files: list[str], ctx: Context ) -> str: """Process multiple files with progress tracking""" total = len(files) results = [] for i, file in enumerate(files): ctx.info(f"Processing {file}") await ctx.report_progress(i, total) # Read resource and process data = await ctx.read_resource(f"file://{file}") results.append(f"Processed {file}") await ctx.report_progress(total, total) return "\n".join(results) # Tool with custom name and description @mcp.tool(name="calculate_bmi", description="Calculate Body Mass Index") def bmi(weight_kg: float, height_m: float) -> float: return weight_kg / (height_m ** 2) ``` -------------------------------- ### Run Type Checking with pyright Source: https://github.com/gwbischof/python-mcp/blob/main/CONTRIBUTING.md Performs static type checking on the project using 'pyright'. This helps catch type-related errors early in the development cycle, improving code quality and maintainability. Adhering to type hints is a project requirement. ```bash uv run pyright ``` -------------------------------- ### Registering MCP Prompts for LLM Interactions Source: https://context7.com/gwbischof/python-mcp/llms.txt Shows how to register reusable prompts for Large Language Models (LLMs) using MCP's `FastMCP` and `base.Message` classes. It covers simple text prompts, multi-message prompts with conversation structure, and prompts with custom names and descriptions. Requires `mcp.server.fastmcp.prompts`. ```python from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp.prompts import base mcp = FastMCP("Assistant") @mcp.prompt() def review_code(code: str) -> str: """Generate a code review prompt""" return f"""Please review this code for: - Bugs and errors - Performance issues - Best practices - Security concerns Code: {code}""" @mcp.prompt() def debug_error(error: str, context: str) -> list[base.Message]: """Create a debugging conversation prompt""" return [ base.UserMessage(f"I'm encountering this error:\n{error}"), base.UserMessage(f"Context:\n{context}"), base.AssistantMessage( "I'll help you debug this. Let me analyze the error and context." ), ] @mcp.prompt(name="sql_helper", description="Help write SQL queries") def sql_assistant(table_schema: str, requirement: str) -> list[base.Message]: """Generate SQL query writing assistance""" return [ base.UserMessage(f"Database schema:\n{table_schema}"), base.UserMessage(f"I need to: {requirement}"), base.AssistantMessage( "I'll help you write the SQL query for this requirement." ), ] ```