From d6a51f70480c7e8135b5dbf995d31344a73e54c9 Mon Sep 17 00:00:00 2001 From: ihrpr Date: Wed, 16 Jul 2025 09:46:47 +0100 Subject: [PATCH 1/2] 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 419e8d351a941367f241270d316d94e5d11a9e05 Mon Sep 17 00:00:00 2001 From: ihrpr Date: Wed, 16 Jul 2025 15:16:23 +0100 Subject: [PATCH 2/2] change an example to remove exceptions and warnings for parsing --- README.md | 67 ++++++++----------- examples/snippets/servers/lowlevel/basic.py | 10 +-- .../snippets/servers/lowlevel/lifespan.py | 7 +- .../servers/lowlevel/structured_output.py | 48 ++++++------- 4 files changed, 57 insertions(+), 75 deletions(-) diff --git a/README.md b/README.md index 16b764ebf..8190f81d6 100644 --- a/README.md +++ b/README.md @@ -1039,12 +1039,7 @@ For more control, you can use the low-level server implementation directly. This ```python -"""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 """ @@ -1160,13 +1155,9 @@ The lifespan API provides: ```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 +uv run examples/snippets/servers/lowlevel/basic.py """ import asyncio @@ -1198,7 +1189,7 @@ async def handle_get_prompt(name: str, arguments: dict[str, str] | None) -> type if name != "example-prompt": raise ValueError(f"Unknown prompt: {name}") - arg1_value = arguments.get("arg1", "default") if arguments else "default" + arg1_value = (arguments or {}).get("arg1", "default") return types.GetPromptResult( description="Example prompt", @@ -1243,11 +1234,7 @@ The low-level server supports structured output for tools, allowing you to retur ```python -"""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 """ @@ -1268,20 +1255,22 @@ async def list_tools() -> list[types.Tool]: """List available tools with structured output schemas.""" return [ types.Tool( - name="calculate", - description="Perform mathematical calculations", + name="get_weather", + description="Get current weather for a city", inputSchema={ "type": "object", - "properties": {"expression": {"type": "string", "description": "Math expression"}}, - "required": ["expression"], + "properties": {"city": {"type": "string", "description": "City name"}}, + "required": ["city"], }, outputSchema={ "type": "object", "properties": { - "result": {"type": "number"}, - "expression": {"type": "string"}, + "temperature": {"type": "number", "description": "Temperature in Celsius"}, + "condition": {"type": "string", "description": "Weather condition"}, + "humidity": {"type": "number", "description": "Humidity percentage"}, + "city": {"type": "string", "description": "City name"}, }, - "required": ["result", "expression"], + "required": ["temperature", "condition", "humidity", "city"], }, ) ] @@ -1290,19 +1279,21 @@ 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: - # 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)}") + if name == "get_weather": + city = arguments["city"] + + # Simulated weather data - in production, call a weather API + weather_data = { + "temperature": 22.5, + "condition": "partly cloudy", + "humidity": 65, + "city": city, # Include the requested city + } + + # low-level server will validate structured output against the tool's + # output schema, and additionally serialize it into a TextContent block + # for backwards compatibility with pre-2025-06-18 clients. + return weather_data else: raise ValueError(f"Unknown tool: {name}") diff --git a/examples/snippets/servers/lowlevel/basic.py b/examples/snippets/servers/lowlevel/basic.py index ca20bf7de..a5c4149df 100644 --- a/examples/snippets/servers/lowlevel/basic.py +++ b/examples/snippets/servers/lowlevel/basic.py @@ -1,10 +1,6 @@ -"""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 +uv run examples/snippets/servers/lowlevel/basic.py """ import asyncio @@ -36,7 +32,7 @@ async def handle_get_prompt(name: str, arguments: dict[str, str] | None) -> type if name != "example-prompt": raise ValueError(f"Unknown prompt: {name}") - arg1_value = arguments.get("arg1", "default") if arguments else "default" + arg1_value = (arguments or {}).get("arg1", "default") return types.GetPromptResult( description="Example prompt", diff --git a/examples/snippets/servers/lowlevel/lifespan.py b/examples/snippets/servers/lowlevel/lifespan.py index 420c67365..61a9fe78e 100644 --- a/examples/snippets/servers/lowlevel/lifespan.py +++ b/examples/snippets/servers/lowlevel/lifespan.py @@ -1,9 +1,4 @@ -"""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 """ diff --git a/examples/snippets/servers/lowlevel/structured_output.py b/examples/snippets/servers/lowlevel/structured_output.py index 8f2b6442d..0237c9ab3 100644 --- a/examples/snippets/servers/lowlevel/structured_output.py +++ b/examples/snippets/servers/lowlevel/structured_output.py @@ -1,8 +1,4 @@ -"""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 """ @@ -23,20 +19,22 @@ async def list_tools() -> list[types.Tool]: """List available tools with structured output schemas.""" return [ types.Tool( - name="calculate", - description="Perform mathematical calculations", + name="get_weather", + description="Get current weather for a city", inputSchema={ "type": "object", - "properties": {"expression": {"type": "string", "description": "Math expression"}}, - "required": ["expression"], + "properties": {"city": {"type": "string", "description": "City name"}}, + "required": ["city"], }, outputSchema={ "type": "object", "properties": { - "result": {"type": "number"}, - "expression": {"type": "string"}, + "temperature": {"type": "number", "description": "Temperature in Celsius"}, + "condition": {"type": "string", "description": "Weather condition"}, + "humidity": {"type": "number", "description": "Humidity percentage"}, + "city": {"type": "string", "description": "City name"}, }, - "required": ["result", "expression"], + "required": ["temperature", "condition", "humidity", "city"], }, ) ] @@ -45,19 +43,21 @@ 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: - # WARNING: eval() is dangerous! Use a safe math parser in production - result = eval(expression) - structured = {"result": result, "expression": expression} + if name == "get_weather": + city = arguments["city"] + + # Simulated weather data - in production, call a weather API + weather_data = { + "temperature": 22.5, + "condition": "partly cloudy", + "humidity": 65, + "city": city, # Include the requested city + } - # 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)}") + # low-level server will validate structured output against the tool's + # output schema, and additionally serialize it into a TextContent block + # for backwards compatibility with pre-2025-06-18 clients. + return weather_data else: raise ValueError(f"Unknown tool: {name}")