From d6a51f70480c7e8135b5dbf995d31344a73e54c9 Mon Sep 17 00:00:00 2001 From: ihrpr Date: Wed, 16 Jul 2025 09:46:47 +0100 Subject: [PATCH 1/7] add lowlevel to snippets --- README.md | 182 +++++++++++++++--- .../snippets/servers/lowlevel/__init__.py | 1 + examples/snippets/servers/lowlevel/basic.py | 70 +++++++ .../snippets/servers/lowlevel/lifespan.py | 108 +++++++++++ .../servers/lowlevel/structured_output.py | 83 ++++++++ 5 files changed, 420 insertions(+), 24 deletions(-) create mode 100644 examples/snippets/servers/lowlevel/__init__.py create mode 100644 examples/snippets/servers/lowlevel/basic.py create mode 100644 examples/snippets/servers/lowlevel/lifespan.py create mode 100644 examples/snippets/servers/lowlevel/structured_output.py diff --git a/README.md b/README.md index c5fb473ca..16b764ebf 100644 --- a/README.md +++ b/README.md @@ -1037,17 +1037,49 @@ For more information on mounting applications in Starlette, see the [Starlette d For more control, you can use the low-level server implementation directly. This gives you full access to the protocol and allows you to customize every aspect of your server, including lifecycle management through the lifespan API: + ```python -from contextlib import asynccontextmanager +"""Low-level server example showing lifespan management. + +This example demonstrates how to use the lifespan API to manage +server startup and shutdown, including resource initialization +and cleanup. + +Run from the repository root: + uv run examples/snippets/servers/lowlevel/lifespan.py +""" + from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +import mcp.server.stdio +import mcp.types as types +from mcp.server.lowlevel import NotificationOptions, Server +from mcp.server.models import InitializationOptions + + +# Mock database class for example +class Database: + """Mock database class for example.""" -from fake_database import Database # Replace with your actual DB type + @classmethod + async def connect(cls) -> "Database": + """Connect to database.""" + print("Database connected") + return cls() -from mcp.server import Server + async def disconnect(self) -> None: + """Disconnect from database.""" + print("Database disconnected") + + async def query(self, query_str: str) -> list[dict[str, str]]: + """Execute a query.""" + # Simulate database query + return [{"id": "1", "name": "Example", "query": query_str}] @asynccontextmanager -async def server_lifespan(server: Server) -> AsyncIterator[dict]: +async def server_lifespan(_server: Server) -> AsyncIterator[dict]: """Manage server startup and shutdown lifecycle.""" # Initialize resources on startup db = await Database.connect() @@ -1062,21 +1094,83 @@ async def server_lifespan(server: Server) -> AsyncIterator[dict]: server = Server("example-server", lifespan=server_lifespan) -# Access lifespan context in handlers +@server.list_tools() +async def handle_list_tools() -> list[types.Tool]: + """List available tools.""" + return [ + types.Tool( + name="query_db", + description="Query the database", + inputSchema={ + "type": "object", + "properties": {"query": {"type": "string", "description": "SQL query to execute"}}, + "required": ["query"], + }, + ) + ] + + @server.call_tool() -async def query_db(name: str, arguments: dict) -> list: +async def query_db(name: str, arguments: dict) -> list[types.TextContent]: + """Handle database query tool call.""" + if name != "query_db": + raise ValueError(f"Unknown tool: {name}") + + # Access lifespan context ctx = server.request_context db = ctx.lifespan_context["db"] - return await db.query(arguments["query"]) + + # Execute query + results = await db.query(arguments["query"]) + + return [types.TextContent(type="text", text=f"Query results: {results}")] + + +async def run(): + """Run the server with lifespan management.""" + async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): + await server.run( + read_stream, + write_stream, + InitializationOptions( + server_name="example-server", + server_version="0.1.0", + capabilities=server.get_capabilities( + notification_options=NotificationOptions(), + experimental_capabilities={}, + ), + ), + ) + + +if __name__ == "__main__": + import asyncio + + asyncio.run(run()) ``` +_Full example: [examples/snippets/servers/lowlevel/lifespan.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/lowlevel/lifespan.py)_ + + The lifespan API provides: - A way to initialize resources when the server starts and clean them up when it stops - Access to initialized resources through the request context in handlers - Type-safe context passing between lifespan and request handlers + ```python +"""Basic low-level server example. + +This example demonstrates the low-level server API with minimal setup, +showing how to implement basic prompts using the raw protocol handlers. + +Run from the repository root: + uv run examples/snippets/servers/lowlevel/basic.py +""" + +import asyncio + import mcp.server.stdio import mcp.types as types from mcp.server.lowlevel import NotificationOptions, Server @@ -1088,38 +1182,37 @@ server = Server("example-server") @server.list_prompts() async def handle_list_prompts() -> list[types.Prompt]: + """List available prompts.""" return [ types.Prompt( name="example-prompt", description="An example prompt template", - arguments=[ - types.PromptArgument( - name="arg1", description="Example argument", required=True - ) - ], + 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: +async def handle_get_prompt(name: str, arguments: dict[str, str] | None) -> types.GetPromptResult: + """Get a specific prompt by name.""" if name != "example-prompt": raise ValueError(f"Unknown prompt: {name}") + arg1_value = arguments.get("arg1", "default") if arguments else "default" + return types.GetPromptResult( description="Example prompt", messages=[ types.PromptMessage( role="user", - content=types.TextContent(type="text", text="Example prompt text"), + content=types.TextContent(type="text", text=f"Example prompt text with argument: {arg1_value}"), ) ], ) async def run(): + """Run the basic low-level server.""" async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, @@ -1136,37 +1229,50 @@ async def run(): if __name__ == "__main__": - import asyncio - asyncio.run(run()) ``` +_Full example: [examples/snippets/servers/lowlevel/basic.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/lowlevel/basic.py)_ + + Caution: The `uv run mcp run` and `uv run mcp dev` tool doesn't support low-level server. #### Structured Output Support The low-level server supports structured output for tools, allowing you to return both human-readable content and machine-readable structured data. Tools can define an `outputSchema` to validate their structured output: + ```python -from types import Any +"""Low-level server example showing structured output support. +This example demonstrates how to use the low-level server API to return +structured data from tools, with automatic validation against output schemas. + +Run from the repository root: + uv run examples/snippets/servers/lowlevel/structured_output.py +""" + +import asyncio +from typing import Any + +import mcp.server.stdio import mcp.types as types -from mcp.server.lowlevel import Server +from mcp.server.lowlevel import NotificationOptions, Server +from mcp.server.models import InitializationOptions server = Server("example-server") @server.list_tools() async def list_tools() -> list[types.Tool]: + """List available tools with structured output schemas.""" return [ types.Tool( name="calculate", description="Perform mathematical calculations", inputSchema={ "type": "object", - "properties": { - "expression": {"type": "string", "description": "Math expression"} - }, + "properties": {"expression": {"type": "string", "description": "Math expression"}}, "required": ["expression"], }, outputSchema={ @@ -1183,10 +1289,12 @@ async def list_tools() -> list[types.Tool]: @server.call_tool() async def call_tool(name: str, arguments: dict[str, Any]) -> dict[str, Any]: + """Handle tool calls with structured output.""" if name == "calculate": expression = arguments["expression"] try: - result = eval(expression) # Use a safe math parser + # WARNING: eval() is dangerous! Use a safe math parser in production + result = eval(expression) structured = {"result": result, "expression": expression} # low-level server will validate structured output against the tool's @@ -1195,8 +1303,34 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> dict[str, Any]: return structured except Exception as e: raise ValueError(f"Calculation error: {str(e)}") + else: + raise ValueError(f"Unknown tool: {name}") + + +async def run(): + """Run the structured output server.""" + async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): + await server.run( + read_stream, + write_stream, + InitializationOptions( + server_name="structured-output-example", + server_version="0.1.0", + capabilities=server.get_capabilities( + notification_options=NotificationOptions(), + experimental_capabilities={}, + ), + ), + ) + + +if __name__ == "__main__": + asyncio.run(run()) ``` +_Full example: [examples/snippets/servers/lowlevel/structured_output.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/lowlevel/structured_output.py)_ + + Tools can return data in three ways: 1. **Content only**: Return a list of content blocks (default behavior before spec revision 2025-06-18) diff --git a/examples/snippets/servers/lowlevel/__init__.py b/examples/snippets/servers/lowlevel/__init__.py new file mode 100644 index 000000000..c6ae62db6 --- /dev/null +++ b/examples/snippets/servers/lowlevel/__init__.py @@ -0,0 +1 @@ +"""Low-level server examples for MCP Python SDK.""" diff --git a/examples/snippets/servers/lowlevel/basic.py b/examples/snippets/servers/lowlevel/basic.py new file mode 100644 index 000000000..ca20bf7de --- /dev/null +++ b/examples/snippets/servers/lowlevel/basic.py @@ -0,0 +1,70 @@ +"""Basic low-level server example. + +This example demonstrates the low-level server API with minimal setup, +showing how to implement basic prompts using the raw protocol handlers. + +Run from the repository root: + uv run examples/snippets/servers/lowlevel/basic.py +""" + +import asyncio + +import mcp.server.stdio +import mcp.types as types +from mcp.server.lowlevel import NotificationOptions, Server +from mcp.server.models import InitializationOptions + +# Create a server instance +server = Server("example-server") + + +@server.list_prompts() +async def handle_list_prompts() -> list[types.Prompt]: + """List available prompts.""" + 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: + """Get a specific prompt by name.""" + if name != "example-prompt": + raise ValueError(f"Unknown prompt: {name}") + + arg1_value = arguments.get("arg1", "default") if arguments else "default" + + return types.GetPromptResult( + description="Example prompt", + messages=[ + types.PromptMessage( + role="user", + content=types.TextContent(type="text", text=f"Example prompt text with argument: {arg1_value}"), + ) + ], + ) + + +async def run(): + """Run the basic low-level server.""" + 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__": + asyncio.run(run()) diff --git a/examples/snippets/servers/lowlevel/lifespan.py b/examples/snippets/servers/lowlevel/lifespan.py new file mode 100644 index 000000000..420c67365 --- /dev/null +++ b/examples/snippets/servers/lowlevel/lifespan.py @@ -0,0 +1,108 @@ +"""Low-level server example showing lifespan management. + +This example demonstrates how to use the lifespan API to manage +server startup and shutdown, including resource initialization +and cleanup. + +Run from the repository root: + uv run examples/snippets/servers/lowlevel/lifespan.py +""" + +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +import mcp.server.stdio +import mcp.types as types +from mcp.server.lowlevel import NotificationOptions, Server +from mcp.server.models import InitializationOptions + + +# Mock database class for example +class Database: + """Mock database class for example.""" + + @classmethod + async def connect(cls) -> "Database": + """Connect to database.""" + print("Database connected") + return cls() + + async def disconnect(self) -> None: + """Disconnect from database.""" + print("Database disconnected") + + async def query(self, query_str: str) -> list[dict[str, str]]: + """Execute a query.""" + # Simulate database query + return [{"id": "1", "name": "Example", "query": query_str}] + + +@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) + + +@server.list_tools() +async def handle_list_tools() -> list[types.Tool]: + """List available tools.""" + return [ + types.Tool( + name="query_db", + description="Query the database", + inputSchema={ + "type": "object", + "properties": {"query": {"type": "string", "description": "SQL query to execute"}}, + "required": ["query"], + }, + ) + ] + + +@server.call_tool() +async def query_db(name: str, arguments: dict) -> list[types.TextContent]: + """Handle database query tool call.""" + if name != "query_db": + raise ValueError(f"Unknown tool: {name}") + + # Access lifespan context + ctx = server.request_context + db = ctx.lifespan_context["db"] + + # Execute query + results = await db.query(arguments["query"]) + + return [types.TextContent(type="text", text=f"Query results: {results}")] + + +async def run(): + """Run the server with lifespan management.""" + async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): + await server.run( + read_stream, + write_stream, + InitializationOptions( + server_name="example-server", + server_version="0.1.0", + capabilities=server.get_capabilities( + notification_options=NotificationOptions(), + experimental_capabilities={}, + ), + ), + ) + + +if __name__ == "__main__": + import asyncio + + asyncio.run(run()) diff --git a/examples/snippets/servers/lowlevel/structured_output.py b/examples/snippets/servers/lowlevel/structured_output.py new file mode 100644 index 000000000..8f2b6442d --- /dev/null +++ b/examples/snippets/servers/lowlevel/structured_output.py @@ -0,0 +1,83 @@ +"""Low-level server example showing structured output support. + +This example demonstrates how to use the low-level server API to return +structured data from tools, with automatic validation against output schemas. + +Run from the repository root: + uv run examples/snippets/servers/lowlevel/structured_output.py +""" + +import asyncio +from typing import Any + +import mcp.server.stdio +import mcp.types as types +from mcp.server.lowlevel import NotificationOptions, Server +from mcp.server.models import InitializationOptions + +server = Server("example-server") + + +@server.list_tools() +async def list_tools() -> list[types.Tool]: + """List available tools with structured output schemas.""" + return [ + types.Tool( + name="calculate", + description="Perform mathematical calculations", + inputSchema={ + "type": "object", + "properties": {"expression": {"type": "string", "description": "Math expression"}}, + "required": ["expression"], + }, + outputSchema={ + "type": "object", + "properties": { + "result": {"type": "number"}, + "expression": {"type": "string"}, + }, + "required": ["result", "expression"], + }, + ) + ] + + +@server.call_tool() +async def call_tool(name: str, arguments: dict[str, Any]) -> dict[str, Any]: + """Handle tool calls with structured output.""" + if name == "calculate": + expression = arguments["expression"] + try: + # WARNING: eval() is dangerous! Use a safe math parser in production + result = eval(expression) + structured = {"result": result, "expression": expression} + + # low-level server will validate structured output against the tool's + # output schema, and automatically serialize it into a TextContent block + # for backwards compatibility with pre-2025-06-18 clients. + return structured + except Exception as e: + raise ValueError(f"Calculation error: {str(e)}") + else: + raise ValueError(f"Unknown tool: {name}") + + +async def run(): + """Run the structured output server.""" + async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): + await server.run( + read_stream, + write_stream, + InitializationOptions( + server_name="structured-output-example", + server_version="0.1.0", + capabilities=server.get_capabilities( + notification_options=NotificationOptions(), + experimental_capabilities={}, + ), + ), + ) + + +if __name__ == "__main__": + asyncio.run(run()) From cb85e20a53369af3de237dcc817479879e23685d Mon Sep 17 00:00:00 2001 From: ihrpr Date: Wed, 16 Jul 2025 12:58:40 +0100 Subject: [PATCH 2/7] streamable http examples --- README.md | 126 +++++++++++------- examples/snippets/clients/streamable_basic.py | 33 +++++ examples/snippets/pyproject.toml | 1 + .../snippets/servers/streamable_config.py | 32 +++++ .../servers/streamable_fastapi_mount.py | 48 +++++++ uv.lock | 20 ++- 6 files changed, 211 insertions(+), 49 deletions(-) create mode 100644 examples/snippets/clients/streamable_basic.py create mode 100644 examples/snippets/servers/streamable_config.py create mode 100644 examples/snippets/servers/streamable_fastapi_mount.py diff --git a/README.md b/README.md index 16b764ebf..66b14e1f8 100644 --- a/README.md +++ b/README.md @@ -863,72 +863,102 @@ Note that `uv run mcp run` or `uv run mcp dev` only supports server using FastMC > **Note**: Streamable HTTP transport is superseding SSE transport for production deployments. + ```python +"""Streamable HTTP server configuration examples. + +This example shows different configuration options for +streamable HTTP servers. + +Run from the repository root: + uv run examples/snippets/servers/streamable_config.py +""" + from mcp.server.fastmcp import FastMCP # Stateful server (maintains session state) mcp = FastMCP("StatefulServer") +# Other configuration options: # Stateless server (no session persistence) -mcp = FastMCP("StatelessServer", stateless_http=True) +# mcp = FastMCP("StatelessServer", stateless_http=True) # Stateless server (no session persistence, no sse stream with supported client) -mcp = FastMCP("StatelessServer", stateless_http=True, json_response=True) +# mcp = FastMCP("StatelessServer", stateless_http=True, json_response=True) + + +# Add a simple tool to demonstrate the server +@mcp.tool() +def greet(name: str = "World") -> str: + """Greet someone by name.""" + return f"Hello, {name}!" + # Run server with streamable_http transport -mcp.run(transport="streamable-http") +if __name__ == "__main__": + mcp.run(transport="streamable-http") ``` +_Full example: [examples/snippets/servers/streamable_config.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/streamable_config.py)_ + + You can mount multiple FastMCP servers in a FastAPI application: + ```python -# echo.py +"""Example of mounting multiple FastMCP servers in a FastAPI application. + +This example shows how to create multiple MCP servers and mount them +at different endpoints in a single FastAPI application. + +Run from the repository root: + uvicorn examples.snippets.servers.streamable_fastapi_mount:app --reload +""" + +import contextlib + +from fastapi import FastAPI + from mcp.server.fastmcp import FastMCP -mcp = FastMCP(name="EchoServer", stateless_http=True) +# Create the Echo server +echo_mcp = FastMCP(name="EchoServer", stateless_http=True) -@mcp.tool() +@echo_mcp.tool() def echo(message: str) -> str: """A simple echo tool""" return f"Echo: {message}" -``` -```python -# math.py -from mcp.server.fastmcp import FastMCP -mcp = FastMCP(name="MathServer", stateless_http=True) +# Create the Math server +math_mcp = FastMCP(name="MathServer", stateless_http=True) -@mcp.tool() +@math_mcp.tool() def add_two(n: int) -> int: """Tool to add two to the input""" 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()) + await stack.enter_async_context(echo_mcp.session_manager.run()) + await stack.enter_async_context(math_mcp.session_manager.run()) yield +# Create the FastAPI app and mount the MCP servers app = FastAPI(lifespan=lifespan) -app.mount("/echo", echo.mcp.streamable_http_app()) -app.mount("/math", math.mcp.streamable_http_app()) +app.mount("/echo", echo_mcp.streamable_http_app()) +app.mount("/math", math_mcp.streamable_http_app()) ``` +_Full example: [examples/snippets/servers/streamable_fastapi_mount.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/streamable_fastapi_mount.py)_ + + For low level server with Streamable HTTP implementations, see: - Stateful server: [`examples/servers/simple-streamablehttp/`](examples/servers/simple-streamablehttp/) @@ -945,26 +975,6 @@ The streamable HTTP transport supports: By default, SSE servers are mounted at `/sse` and Streamable HTTP servers are mounted at `/mcp`. You can customize these paths using the methods described below. -#### Streamable HTTP servers - -The following example shows how to use `streamable_http_app()`, a method that returns a `Starlette` application object. -You can then append additional routes to that application as needed. - -```python -mcp = FastMCP("My App") - -app = mcp.streamable_http_app() -# Additional non-MCP routes can be added like so: -# from starlette.routing import Route -# app.router.routes.append(Route("/", endpoint=other_route_function)) -``` - -To customize the route from the default of "/mcp", either specify the `streamable_http_path` option for the `FastMCP` constructor, -or set `FASTMCP_STREAMABLE_HTTP_PATH` environment variable. - -Note that in Starlette and FastAPI (which is based on Starlette), the "/mcp" route will redirect to "/mcp/", -so you may need to use "/mcp/" when pointing MCP clients at your servers. - For more information on mounting applications in Starlette, see the [Starlette documentation](https://www.starlette.io/routing/#submounting-routes). #### SSE servers @@ -1437,14 +1447,26 @@ _Full example: [examples/snippets/clients/stdio_client.py](https://github.com/mo Clients can also connect using [Streamable HTTP transport](https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#streamable-http): + ```python -from mcp.client.streamable_http import streamablehttp_client +"""Basic streamable HTTP client example. + +This example shows the minimal code needed to connect to +a streamable HTTP server and call a tool. + +Run from the repository root: + uv run examples/snippets/clients/streamable_basic.py +""" + +import asyncio + from mcp import ClientSession +from mcp.client.streamable_http import streamablehttp_client async def main(): # Connect to a streamable HTTP server - async with streamablehttp_client("example/mcp") as ( + async with streamablehttp_client("http://localhost:8000/mcp") as ( read_stream, write_stream, _, @@ -1453,10 +1475,18 @@ async def main(): 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"}) + # List available tools + tools = await session.list_tools() + print(f"Available tools: {[tool.name for tool in tools.tools]}") + + +if __name__ == "__main__": + asyncio.run(main()) ``` +_Full example: [examples/snippets/clients/streamable_basic.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/streamable_basic.py)_ + + ### Client Display Utilities When building MCP clients, the SDK provides utilities to help display human-readable names for tools, resources, and prompts: diff --git a/examples/snippets/clients/streamable_basic.py b/examples/snippets/clients/streamable_basic.py new file mode 100644 index 000000000..aa045c3ba --- /dev/null +++ b/examples/snippets/clients/streamable_basic.py @@ -0,0 +1,33 @@ +"""Basic streamable HTTP client example. + +This example shows the minimal code needed to connect to +a streamable HTTP server and call a tool. + +Run from the repository root: + uv run examples/snippets/clients/streamable_basic.py +""" + +import asyncio + +from mcp import ClientSession +from mcp.client.streamable_http import streamablehttp_client + + +async def main(): + # Connect to a streamable HTTP server + async with streamablehttp_client("http://localhost:8000/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() + # List available tools + tools = await session.list_tools() + print(f"Available tools: {[tool.name for tool in tools.tools]}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/snippets/pyproject.toml b/examples/snippets/pyproject.toml index 7e6e032c7..bbbcee65f 100644 --- a/examples/snippets/pyproject.toml +++ b/examples/snippets/pyproject.toml @@ -5,6 +5,7 @@ description = "MCP Example Snippets" requires-python = ">=3.10" dependencies = [ "mcp", + "fastapi", ] [build-system] diff --git a/examples/snippets/servers/streamable_config.py b/examples/snippets/servers/streamable_config.py new file mode 100644 index 000000000..0d3d4bf65 --- /dev/null +++ b/examples/snippets/servers/streamable_config.py @@ -0,0 +1,32 @@ +"""Streamable HTTP server configuration examples. + +This example shows different configuration options for +streamable HTTP servers. + +Run from the repository root: + uv run examples/snippets/servers/streamable_config.py +""" + +from mcp.server.fastmcp import FastMCP + +# Stateful server (maintains session state) +mcp = FastMCP("StatefulServer") + +# Other configuration options: +# Stateless server (no session persistence) +# mcp = FastMCP("StatelessServer", stateless_http=True) + +# Stateless server (no session persistence, no sse stream with supported client) +# mcp = FastMCP("StatelessServer", stateless_http=True, json_response=True) + + +# Add a simple tool to demonstrate the server +@mcp.tool() +def greet(name: str = "World") -> str: + """Greet someone by name.""" + return f"Hello, {name}!" + + +# Run server with streamable_http transport +if __name__ == "__main__": + mcp.run(transport="streamable-http") diff --git a/examples/snippets/servers/streamable_fastapi_mount.py b/examples/snippets/servers/streamable_fastapi_mount.py new file mode 100644 index 000000000..2a1760e22 --- /dev/null +++ b/examples/snippets/servers/streamable_fastapi_mount.py @@ -0,0 +1,48 @@ +"""Example of mounting multiple FastMCP servers in a FastAPI application. + +This example shows how to create multiple MCP servers and mount them +at different endpoints in a single FastAPI application. + +Run from the repository root: + uvicorn examples.snippets.servers.streamable_fastapi_mount:app --reload +""" + +import contextlib + +from fastapi import FastAPI + +from mcp.server.fastmcp import FastMCP + +# Create the Echo server +echo_mcp = FastMCP(name="EchoServer", stateless_http=True) + + +@echo_mcp.tool() +def echo(message: str) -> str: + """A simple echo tool""" + return f"Echo: {message}" + + +# Create the Math server +math_mcp = FastMCP(name="MathServer", stateless_http=True) + + +@math_mcp.tool() +def add_two(n: int) -> int: + """Tool to add two to the input""" + return n + 2 + + +# 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 + + +# Create the FastAPI app and mount the MCP servers +app = FastAPI(lifespan=lifespan) +app.mount("/echo", echo_mcp.streamable_http_app()) +app.mount("/math", math_mcp.streamable_http_app()) diff --git a/uv.lock b/uv.lock index 7a34275ce..76f7aea0f 100644 --- a/uv.lock +++ b/uv.lock @@ -350,6 +350,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/7b/8f/c4d9bafc34ad7ad5d8dc16dd1347ee0e507a52c3adb6bfa8887e1c6a26ba/executing-2.2.0-py2.py3-none-any.whl", hash = "sha256:11387150cad388d62750327a53d3339fad4888b39a6fe233c3afbb54ecffd3aa", size = 26702, upload-time = "2025-01-22T15:41:25.929Z" }, ] +[[package]] +name = "fastapi" +version = "0.116.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pydantic" }, + { name = "starlette" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/78/d7/6c8b3bfe33eeffa208183ec037fee0cce9f7f024089ab1c5d12ef04bd27c/fastapi-0.116.1.tar.gz", hash = "sha256:ed52cbf946abfd70c5a0dccb24673f0670deeb517a88b3544d03c2a6bf283143", size = 296485, upload-time = "2025-07-11T16:22:32.057Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e5/47/d63c60f59a59467fda0f93f46335c9d18526d7071f025cb5b89d5353ea42/fastapi-0.116.1-py3-none-any.whl", hash = "sha256:c46ac7c312df840f0c9e220f7964bada936781bc4e2e6eb71f1c4d7553786565", size = 95631, upload-time = "2025-07-11T16:22:30.485Z" }, +] + [[package]] name = "ghp-import" version = "2.1.0" @@ -880,11 +894,15 @@ name = "mcp-snippets" version = "0.1.0" source = { editable = "examples/snippets" } dependencies = [ + { name = "fastapi" }, { name = "mcp" }, ] [package.metadata] -requires-dist = [{ name = "mcp", editable = "." }] +requires-dist = [ + { name = "fastapi" }, + { name = "mcp", editable = "." }, +] [[package]] name = "mdurl" From cb1a316032d1cf96191870d6956e1ffc28bd8b5c Mon Sep 17 00:00:00 2001 From: ihrpr Date: Wed, 16 Jul 2025 13:52:24 +0100 Subject: [PATCH 3/7] pyright --- .pre-commit-config.yaml | 7 +++++++ CONTRIBUTING.md | 6 ++++++ 2 files changed, 13 insertions(+) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 553c52d62..c27b9fc94 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -44,6 +44,13 @@ repos: language: system types: [python] pass_filenames: false + exclude: ^examples/snippets/ + - id: pyright-snippets + name: pyright (snippets) + entry: bash -c "cd examples/snippets && uv run pyright ." + language: system + files: ^examples/snippets/.*\.py$ + pass_filenames: false - id: uv-lock-check name: Check uv.lock is up to date entry: uv lock --check diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index fe78cf25c..612d96050 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -37,6 +37,12 @@ uv run pytest uv run pyright ``` +Note: The `examples/snippets` directory has its own pyright configuration and is checked separately by pre-commit. If you're modifying examples in that directory, you can test locally with: + +```bash +cd examples/snippets && uv run pyright +``` + 6. Run linting: ```bash From 925b22f572a193a8aa036f495248ee659a00f82c Mon Sep 17 00:00:00 2001 From: ihrpr Date: Wed, 16 Jul 2025 14:21:26 +0100 Subject: [PATCH 4/7] revert --- .pre-commit-config.yaml | 7 ------- CONTRIBUTING.md | 6 ------ 2 files changed, 13 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index c27b9fc94..553c52d62 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -44,13 +44,6 @@ repos: language: system types: [python] pass_filenames: false - exclude: ^examples/snippets/ - - id: pyright-snippets - name: pyright (snippets) - entry: bash -c "cd examples/snippets && uv run pyright ." - language: system - files: ^examples/snippets/.*\.py$ - pass_filenames: false - id: uv-lock-check name: Check uv.lock is up to date entry: uv lock --check diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 612d96050..fe78cf25c 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -37,12 +37,6 @@ uv run pytest uv run pyright ``` -Note: The `examples/snippets` directory has its own pyright configuration and is checked separately by pre-commit. If you're modifying examples in that directory, you can test locally with: - -```bash -cd examples/snippets && uv run pyright -``` - 6. Run linting: ```bash From 362b00ec2deb91b9982f9829e458d34f0c8cf193 Mon Sep 17 00:00:00 2001 From: ihrpr Date: Wed, 16 Jul 2025 14:48:07 +0100 Subject: [PATCH 5/7] use starlette --- README.md | 29 +++++++++++-------- examples/snippets/pyproject.toml | 1 - ...mount.py => streamable_starlette_mount.py} | 23 +++++++++------ uv.lock | 20 +------------ 4 files changed, 32 insertions(+), 41 deletions(-) rename examples/snippets/servers/{streamable_fastapi_mount.py => streamable_starlette_mount.py} (61%) diff --git a/README.md b/README.md index 66b14e1f8..785051ba3 100644 --- a/README.md +++ b/README.md @@ -902,22 +902,23 @@ if __name__ == "__main__": _Full example: [examples/snippets/servers/streamable_config.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/streamable_config.py)_ -You can mount multiple FastMCP servers in a FastAPI application: +You can mount multiple FastMCP servers in a Starlette application: - + ```python -"""Example of mounting multiple FastMCP servers in a FastAPI application. +"""Example of mounting multiple FastMCP servers in a Starlette application. This example shows how to create multiple MCP servers and mount them -at different endpoints in a single FastAPI application. +at different endpoints in a single Starlette application. Run from the repository root: - uvicorn examples.snippets.servers.streamable_fastapi_mount:app --reload + uvicorn examples.snippets.servers.streamable_starlette_mount:app --reload """ import contextlib -from fastapi import FastAPI +from starlette.applications import Starlette +from starlette.routing import Mount from mcp.server.fastmcp import FastMCP @@ -943,20 +944,24 @@ def add_two(n: int) -> int: # Create a combined lifespan to manage both session managers @contextlib.asynccontextmanager -async def lifespan(app: FastAPI): +async def lifespan(app): 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 -# Create the FastAPI app and mount the MCP servers -app = FastAPI(lifespan=lifespan) -app.mount("/echo", echo_mcp.streamable_http_app()) -app.mount("/math", math_mcp.streamable_http_app()) +# Create the Starlette app and mount the MCP servers +app = Starlette( + routes=[ + Mount("/echo", echo_mcp.streamable_http_app()), + Mount("/math", math_mcp.streamable_http_app()), + ], + lifespan=lifespan, +) ``` -_Full example: [examples/snippets/servers/streamable_fastapi_mount.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/streamable_fastapi_mount.py)_ +_Full example: [examples/snippets/servers/streamable_starlette_mount.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/streamable_starlette_mount.py)_ For low level server with Streamable HTTP implementations, see: diff --git a/examples/snippets/pyproject.toml b/examples/snippets/pyproject.toml index bbbcee65f..7e6e032c7 100644 --- a/examples/snippets/pyproject.toml +++ b/examples/snippets/pyproject.toml @@ -5,7 +5,6 @@ description = "MCP Example Snippets" requires-python = ">=3.10" dependencies = [ "mcp", - "fastapi", ] [build-system] diff --git a/examples/snippets/servers/streamable_fastapi_mount.py b/examples/snippets/servers/streamable_starlette_mount.py similarity index 61% rename from examples/snippets/servers/streamable_fastapi_mount.py rename to examples/snippets/servers/streamable_starlette_mount.py index 2a1760e22..b19ec7b1c 100644 --- a/examples/snippets/servers/streamable_fastapi_mount.py +++ b/examples/snippets/servers/streamable_starlette_mount.py @@ -1,15 +1,16 @@ -"""Example of mounting multiple FastMCP servers in a FastAPI application. +"""Example of mounting multiple FastMCP servers in a Starlette application. This example shows how to create multiple MCP servers and mount them -at different endpoints in a single FastAPI application. +at different endpoints in a single Starlette application. Run from the repository root: - uvicorn examples.snippets.servers.streamable_fastapi_mount:app --reload + uvicorn examples.snippets.servers.streamable_starlette_mount:app --reload """ import contextlib -from fastapi import FastAPI +from starlette.applications import Starlette +from starlette.routing import Mount from mcp.server.fastmcp import FastMCP @@ -35,14 +36,18 @@ def add_two(n: int) -> int: # Create a combined lifespan to manage both session managers @contextlib.asynccontextmanager -async def lifespan(app: FastAPI): +async def lifespan(app): 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 -# Create the FastAPI app and mount the MCP servers -app = FastAPI(lifespan=lifespan) -app.mount("/echo", echo_mcp.streamable_http_app()) -app.mount("/math", math_mcp.streamable_http_app()) +# Create the Starlette app and mount the MCP servers +app = Starlette( + routes=[ + Mount("/echo", echo_mcp.streamable_http_app()), + Mount("/math", math_mcp.streamable_http_app()), + ], + lifespan=lifespan, +) diff --git a/uv.lock b/uv.lock index 76f7aea0f..7a34275ce 100644 --- a/uv.lock +++ b/uv.lock @@ -350,20 +350,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/7b/8f/c4d9bafc34ad7ad5d8dc16dd1347ee0e507a52c3adb6bfa8887e1c6a26ba/executing-2.2.0-py2.py3-none-any.whl", hash = "sha256:11387150cad388d62750327a53d3339fad4888b39a6fe233c3afbb54ecffd3aa", size = 26702, upload-time = "2025-01-22T15:41:25.929Z" }, ] -[[package]] -name = "fastapi" -version = "0.116.1" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "pydantic" }, - { name = "starlette" }, - { name = "typing-extensions" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/78/d7/6c8b3bfe33eeffa208183ec037fee0cce9f7f024089ab1c5d12ef04bd27c/fastapi-0.116.1.tar.gz", hash = "sha256:ed52cbf946abfd70c5a0dccb24673f0670deeb517a88b3544d03c2a6bf283143", size = 296485, upload-time = "2025-07-11T16:22:32.057Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/e5/47/d63c60f59a59467fda0f93f46335c9d18526d7071f025cb5b89d5353ea42/fastapi-0.116.1-py3-none-any.whl", hash = "sha256:c46ac7c312df840f0c9e220f7964bada936781bc4e2e6eb71f1c4d7553786565", size = 95631, upload-time = "2025-07-11T16:22:30.485Z" }, -] - [[package]] name = "ghp-import" version = "2.1.0" @@ -894,15 +880,11 @@ name = "mcp-snippets" version = "0.1.0" source = { editable = "examples/snippets" } dependencies = [ - { name = "fastapi" }, { name = "mcp" }, ] [package.metadata] -requires-dist = [ - { name = "fastapi" }, - { name = "mcp", editable = "." }, -] +requires-dist = [{ name = "mcp", editable = "." }] [[package]] name = "mdurl" From d87f26d0a75fcece7a60b643a2e125dbee4d650c Mon Sep 17 00:00:00 2001 From: ihrpr Date: Wed, 16 Jul 2025 16:09:54 +0100 Subject: [PATCH 6/7] remove docstring --- README.md | 32 ++++--------------- .../snippets/clients/completion_client.py | 4 +-- .../snippets/clients/display_utilities.py | 6 +--- examples/snippets/clients/stdio_client.py | 4 +-- examples/snippets/clients/streamable_basic.py | 6 +--- .../snippets/servers/streamable_config.py | 6 +--- .../servers/streamable_starlette_mount.py | 6 +--- 7 files changed, 12 insertions(+), 52 deletions(-) diff --git a/README.md b/README.md index 31cbbfdb5..d8f8eee2c 100644 --- a/README.md +++ b/README.md @@ -532,9 +532,7 @@ Client usage: ```python -"""MCP client example showing completion usage. - -This example demonstrates how to use the completion feature in MCP clients. +""" cd to the `examples/snippets` directory and run: uv run completion-client """ @@ -865,11 +863,7 @@ Note that `uv run mcp run` or `uv run mcp dev` only supports server using FastMC ```python -"""Streamable HTTP server configuration examples. - -This example shows different configuration options for -streamable HTTP servers. - +""" Run from the repository root: uv run examples/snippets/servers/streamable_config.py """ @@ -906,11 +900,7 @@ You can mount multiple FastMCP servers in a Starlette application: ```python -"""Example of mounting multiple FastMCP servers in a Starlette application. - -This example shows how to create multiple MCP servers and mount them -at different endpoints in a single Starlette application. - +""" Run from the repository root: uvicorn examples.snippets.servers.streamable_starlette_mount:app --reload """ @@ -1351,9 +1341,7 @@ The SDK provides a high-level client interface for connecting to MCP servers usi ```python -"""MCP client example using stdio transport. - -This is a documentation example showing how to write an MCP client. +""" cd to the `examples/snippets/clients` directory and run: uv run client """ @@ -1445,11 +1433,7 @@ Clients can also connect using [Streamable HTTP transport](https://modelcontextp ```python -"""Basic streamable HTTP client example. - -This example shows the minimal code needed to connect to -a streamable HTTP server and call a tool. - +""" Run from the repository root: uv run examples/snippets/clients/streamable_basic.py """ @@ -1489,11 +1473,7 @@ When building MCP clients, the SDK provides utilities to help display human-read ```python -"""Client display utilities example. - -This example shows how to use the SDK's display utilities to show -human-readable names for tools, resources, and prompts. - +""" cd to the `examples/snippets` directory and run: uv run display-utilities-client """ diff --git a/examples/snippets/clients/completion_client.py b/examples/snippets/clients/completion_client.py index d8745ea1e..8c5615926 100644 --- a/examples/snippets/clients/completion_client.py +++ b/examples/snippets/clients/completion_client.py @@ -1,6 +1,4 @@ -"""MCP client example showing completion usage. - -This example demonstrates how to use the completion feature in MCP clients. +""" cd to the `examples/snippets` directory and run: uv run completion-client """ diff --git a/examples/snippets/clients/display_utilities.py b/examples/snippets/clients/display_utilities.py index ee2280b3a..5f1d50510 100644 --- a/examples/snippets/clients/display_utilities.py +++ b/examples/snippets/clients/display_utilities.py @@ -1,8 +1,4 @@ -"""Client display utilities example. - -This example shows how to use the SDK's display utilities to show -human-readable names for tools, resources, and prompts. - +""" cd to the `examples/snippets` directory and run: uv run display-utilities-client """ diff --git a/examples/snippets/clients/stdio_client.py b/examples/snippets/clients/stdio_client.py index 22ad933ad..74a6f09df 100644 --- a/examples/snippets/clients/stdio_client.py +++ b/examples/snippets/clients/stdio_client.py @@ -1,6 +1,4 @@ -"""MCP client example using stdio transport. - -This is a documentation example showing how to write an MCP client. +""" cd to the `examples/snippets/clients` directory and run: uv run client """ diff --git a/examples/snippets/clients/streamable_basic.py b/examples/snippets/clients/streamable_basic.py index aa045c3ba..108439613 100644 --- a/examples/snippets/clients/streamable_basic.py +++ b/examples/snippets/clients/streamable_basic.py @@ -1,8 +1,4 @@ -"""Basic streamable HTTP client example. - -This example shows the minimal code needed to connect to -a streamable HTTP server and call a tool. - +""" Run from the repository root: uv run examples/snippets/clients/streamable_basic.py """ diff --git a/examples/snippets/servers/streamable_config.py b/examples/snippets/servers/streamable_config.py index 0d3d4bf65..e265f6381 100644 --- a/examples/snippets/servers/streamable_config.py +++ b/examples/snippets/servers/streamable_config.py @@ -1,8 +1,4 @@ -"""Streamable HTTP server configuration examples. - -This example shows different configuration options for -streamable HTTP servers. - +""" Run from the repository root: uv run examples/snippets/servers/streamable_config.py """ diff --git a/examples/snippets/servers/streamable_starlette_mount.py b/examples/snippets/servers/streamable_starlette_mount.py index b19ec7b1c..4e12b496d 100644 --- a/examples/snippets/servers/streamable_starlette_mount.py +++ b/examples/snippets/servers/streamable_starlette_mount.py @@ -1,8 +1,4 @@ -"""Example of mounting multiple FastMCP servers in a Starlette application. - -This example shows how to create multiple MCP servers and mount them -at different endpoints in a single Starlette application. - +""" Run from the repository root: uvicorn examples.snippets.servers.streamable_starlette_mount:app --reload """ From 50d124842408a4afda68cff15b2a80ce75989483 Mon Sep 17 00:00:00 2001 From: ihrpr Date: Wed, 16 Jul 2025 21:33:28 +0100 Subject: [PATCH 7/7] add type --- README.md | 2 +- examples/snippets/servers/streamable_starlette_mount.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index d8f8eee2c..2cb90770c 100644 --- a/README.md +++ b/README.md @@ -934,7 +934,7 @@ def add_two(n: int) -> int: # Create a combined lifespan to manage both session managers @contextlib.asynccontextmanager -async def lifespan(app): +async def lifespan(app: Starlette): 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()) diff --git a/examples/snippets/servers/streamable_starlette_mount.py b/examples/snippets/servers/streamable_starlette_mount.py index 4e12b496d..19e41294b 100644 --- a/examples/snippets/servers/streamable_starlette_mount.py +++ b/examples/snippets/servers/streamable_starlette_mount.py @@ -32,7 +32,7 @@ def add_two(n: int) -> int: # Create a combined lifespan to manage both session managers @contextlib.asynccontextmanager -async def lifespan(app): +async def lifespan(app: Starlette): 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())