diff --git a/README.md b/README.md index 3cad2aade..08ed251cc 100644 --- a/README.md +++ b/README.md @@ -41,9 +41,6 @@ - [Claude Desktop Integration](#claude-desktop-integration) - [Direct Execution](#direct-execution) - [Mounting to an Existing ASGI Server](#mounting-to-an-existing-asgi-server) - - [Examples](#examples) - - [Echo Server](#echo-server) - - [SQLite Explorer](#sqlite-explorer) - [Advanced Usage](#advanced-usage) - [Low-Level Server](#low-level-server) - [Writing MCP Clients](#writing-mcp-clients) @@ -112,8 +109,15 @@ uv run mcp Let's create a simple MCP server that exposes a calculator tool and some data: + ```python -# server.py +""" +FastMCP quickstart example. + +cd to the `examples/snippets/clients` directory and run: + uv run server fastmcp_quickstart stdio +""" + from mcp.server.fastmcp import FastMCP # Create an MCP server @@ -122,7 +126,7 @@ mcp = FastMCP("Demo") # Add an addition tool @mcp.tool() -def sum(a: int, b: int) -> int: +def add(a: int, b: int) -> int: """Add two numbers""" return a + b @@ -132,8 +136,24 @@ def sum(a: int, b: int) -> int: def get_greeting(name: str) -> str: """Get a personalized greeting""" return f"Hello, {name}!" + + +# Add a prompt +@mcp.prompt() +def greet_user(name: str, style: str = "friendly") -> str: + """Generate a greeting prompt""" + styles = { + "friendly": "Please write a warm, friendly greeting", + "formal": "Please write a formal, professional greeting", + "casual": "Please write a casual, relaxed greeting", + } + + return f"{styles.get(style, styles['friendly'])} for someone named {name}." ``` +_Full example: [examples/snippets/servers/fastmcp_quickstart.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/fastmcp_quickstart.py)_ + + You can install this server in [Claude Desktop](https://claude.ai/download) and interact with it right away by running: ```bash @@ -161,31 +181,45 @@ The [Model Context Protocol (MCP)](https://modelcontextprotocol.io) lets you bui The FastMCP server is your core interface to the MCP protocol. It handles connection management, protocol compliance, and message routing: + ```python -# Add lifespan support for startup/shutdown with strong typing -from contextlib import asynccontextmanager +"""Example showing lifespan support for startup/shutdown with strong typing.""" + from collections.abc import AsyncIterator +from contextlib import asynccontextmanager from dataclasses import dataclass -from fake_database import Database # Replace with your actual DB type +from mcp.server.fastmcp import Context, FastMCP -from mcp.server.fastmcp import FastMCP -# Create a named server -mcp = FastMCP("My App") +# Mock database class for example +class Database: + """Mock database class for example.""" + + @classmethod + async def connect(cls) -> "Database": + """Connect to database.""" + return cls() + + async def disconnect(self) -> None: + """Disconnect from database.""" + pass -# Specify dependencies for deployment and development -mcp = FastMCP("My App", dependencies=["pandas", "numpy"]) + def query(self) -> str: + """Execute a query.""" + return "Query result" @dataclass class AppContext: + """Application context with typed dependencies.""" + db: Database @asynccontextmanager async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]: - """Manage application lifecycle with type-safe context""" + """Manage application lifecycle with type-safe context.""" # Initialize on startup db = await Database.connect() try: @@ -201,13 +235,15 @@ mcp = FastMCP("My App", lifespan=app_lifespan) # Access type-safe lifespan context in tools @mcp.tool() -def query_db() -> str: - """Tool that uses initialized resources""" - ctx = mcp.get_context() +def query_db(ctx: Context) -> str: + """Tool that uses initialized resources.""" db = ctx.request_context.lifespan_context.db return db.query() ``` +_Full example: [examples/snippets/servers/lifespan_example.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/lifespan_example.py)_ + + ### Resources Resources are how you expose data to LLMs. They're similar to GET endpoints in a REST API - they provide data but shouldn't perform significant computation or have side effects: @@ -298,16 +334,21 @@ causes the tool to be classified as structured _and this is undesirable_, the classification can be suppressed by passing `structured_output=False` to the `@tool` decorator. + ```python -from mcp.server.fastmcp import FastMCP +"""Example showing structured output with tools.""" + from pydantic import BaseModel, Field -from typing import TypedDict -mcp = FastMCP("Weather Service") +from mcp.server.fastmcp import FastMCP + +mcp = FastMCP("Structured Output Example") # Using Pydantic models for rich structured data class WeatherData(BaseModel): + """Weather information structure.""" + temperature: float = Field(description="Temperature in Celsius") humidity: float = Field(description="Humidity percentage") condition: str @@ -316,12 +357,20 @@ class WeatherData(BaseModel): @mcp.tool() def get_weather(city: str) -> WeatherData: - """Get structured weather data""" + """Get weather for a city - returns structured data.""" + # Simulated weather data return WeatherData( - temperature=22.5, humidity=65.0, condition="partly cloudy", wind_speed=12.3 + temperature=72.5, + humidity=45.0, + condition="sunny", + wind_speed=5.2, ) +``` +_Full example: [examples/snippets/servers/structured_output.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/structured_output.py)_ + +```python # Using TypedDict for simpler structures class LocationInfo(TypedDict): latitude: float @@ -421,11 +470,15 @@ _Full example: [examples/snippets/servers/basic_prompt.py](https://github.com/mo FastMCP provides an `Image` class that automatically handles image data: + ```python -from mcp.server.fastmcp import FastMCP, Image +"""Example showing image handling with FastMCP.""" + from PIL import Image as PILImage -mcp = FastMCP("My App") +from mcp.server.fastmcp import FastMCP, Image + +mcp = FastMCP("Image Example") @mcp.tool() @@ -436,6 +489,9 @@ def create_thumbnail(image_path: str) -> Image: return Image(data=img.tobytes(), format="png") ``` +_Full example: [examples/snippets/servers/images.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/images.py)_ + + ### Context The Context object gives your tools and resources access to MCP capabilities: @@ -473,86 +529,91 @@ MCP supports providing completion suggestions for prompt arguments and resource Client usage: + ```python -from mcp.client.session import ClientSession -from mcp.types import ResourceTemplateReference +"""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 +""" -async def use_completion(session: ClientSession): - # Complete without context - result = await session.complete( - ref=ResourceTemplateReference( - type="ref/resource", uri="github://repos/{owner}/{repo}" - ), - argument={"name": "owner", "value": "model"}, - ) - - # Complete with context - repo suggestions based on owner - result = await session.complete( - ref=ResourceTemplateReference( - type="ref/resource", uri="github://repos/{owner}/{repo}" - ), - argument={"name": "repo", "value": "test"}, - context_arguments={"owner": "modelcontextprotocol"}, - ) -``` +import asyncio +import os -Server implementation: +from mcp import ClientSession, StdioServerParameters +from mcp.client.stdio import stdio_client +from mcp.types import PromptReference, ResourceTemplateReference - -```python -from mcp.server.fastmcp import FastMCP -from mcp.types import ( - Completion, - CompletionArgument, - CompletionContext, - PromptReference, - ResourceTemplateReference, +# Create server parameters for stdio connection +server_params = StdioServerParameters( + command="uv", # Using uv to run the server + args=["run", "server", "completion", "stdio"], # Server with completion support + env={"UV_INDEX": os.environ.get("UV_INDEX", "")}, ) -mcp = FastMCP(name="Example") +async def run(): + """Run the completion client example.""" + async with stdio_client(server_params) as (read, write): + async with ClientSession(read, write) as session: + # Initialize the connection + await session.initialize() + + # List available resource templates + templates = await session.list_resource_templates() + print("Available resource templates:") + for template in templates.resourceTemplates: + print(f" - {template.uriTemplate}") -@mcp.resource("github://repos/{owner}/{repo}") -def github_repo(owner: str, repo: str) -> str: - """GitHub repository resource.""" - return f"Repository: {owner}/{repo}" + # List available prompts + prompts = await session.list_prompts() + print("\nAvailable prompts:") + for prompt in prompts.prompts: + print(f" - {prompt.name}") + + # Complete resource template arguments + if templates.resourceTemplates: + template = templates.resourceTemplates[0] + print(f"\nCompleting arguments for resource template: {template.uriTemplate}") + + # Complete without context + result = await session.complete( + ref=ResourceTemplateReference(type="ref/resource", uri=template.uriTemplate), + argument={"name": "owner", "value": "model"}, + ) + print(f"Completions for 'owner' starting with 'model': {result.completion.values}") + # Complete with context - repo suggestions based on owner + result = await session.complete( + ref=ResourceTemplateReference(type="ref/resource", uri=template.uriTemplate), + argument={"name": "repo", "value": ""}, + context_arguments={"owner": "modelcontextprotocol"}, + ) + print(f"Completions for 'repo' with owner='modelcontextprotocol': {result.completion.values}") -@mcp.prompt(description="Code review prompt") -def review_code(language: str, code: str) -> str: - """Generate a code review.""" - return f"Review this {language} code:\n{code}" + # Complete prompt arguments + if prompts.prompts: + prompt_name = prompts.prompts[0].name + print(f"\nCompleting arguments for prompt: {prompt_name}") + result = await session.complete( + ref=PromptReference(type="ref/prompt", name=prompt_name), + argument={"name": "style", "value": ""}, + ) + print(f"Completions for 'style' argument: {result.completion.values}") -@mcp.completion() -async def handle_completion( - ref: PromptReference | ResourceTemplateReference, - argument: CompletionArgument, - context: CompletionContext | None, -) -> Completion | None: - """Provide completions for prompts and resources.""" - # Complete programming languages for the prompt - if isinstance(ref, PromptReference): - if ref.name == "review_code" and argument.name == "language": - languages = ["python", "javascript", "typescript", "go", "rust"] - return Completion( - values=[lang for lang in languages if lang.startswith(argument.value)], - hasMore=False, - ) +def main(): + """Entry point for the completion client.""" + asyncio.run(run()) - # Complete repository names for GitHub resources - if isinstance(ref, ResourceTemplateReference): - if ref.uri == "github://repos/{owner}/{repo}" and argument.name == "repo": - if context and context.arguments and context.arguments.get("owner") == "modelcontextprotocol": - repos = ["python-sdk", "typescript-sdk", "specification"] - return Completion(values=repos, hasMore=False) - return None +if __name__ == "__main__": + main() ``` -_Full example: [examples/snippets/servers/completion.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/completion.py)_ +_Full example: [examples/snippets/clients/completion_client.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/completion_client.py)_ ### Elicitation @@ -920,67 +981,6 @@ if __name__ == "__main__": For more information on mounting applications in Starlette, see the [Starlette documentation](https://www.starlette.io/routing/#submounting-routes). -## Examples - -### Echo Server - -A simple server demonstrating resources, tools, and prompts: - -```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}" -``` - -### SQLite Explorer - -A more complex example showing database integration: - -```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)}" -``` - ## Advanced Usage ### Low-Level Server @@ -1159,22 +1159,37 @@ When an `outputSchema` is defined, the server automatically validates the struct The SDK provides a high-level client interface for connecting to MCP servers using various [transports](https://modelcontextprotocol.io/specification/2025-03-26/basic/transports): + ```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 +""" + +import asyncio +import os + +from pydantic import AnyUrl + from mcp import ClientSession, StdioServerParameters, types from mcp.client.stdio import stdio_client +from mcp.shared.context import RequestContext # 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 + command="uv", # Using uv to run the server + args=["run", "server", "fastmcp_quickstart", "stdio"], # We're already in snippets dir + env={"UV_INDEX": os.environ.get("UV_INDEX", "")}, ) # Optional: create a sampling callback async def handle_sampling_message( - message: types.CreateMessageRequestParams, + context: RequestContext, params: types.CreateMessageRequestParams ) -> types.CreateMessageResult: + print(f"Sampling request: {params.messages}") return types.CreateMessageResult( role="assistant", content=types.TextContent( @@ -1188,39 +1203,54 @@ async def handle_sampling_message( async def run(): async with stdio_client(server_params) as (read, write): - async with ClientSession( - read, write, sampling_callback=handle_sampling_message - ) as session: + 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() + print(f"Available prompts: {[p.name for p in prompts.prompts]}") - # Get a prompt - prompt = await session.get_prompt( - "example-prompt", arguments={"arg1": "value"} - ) + # Get a prompt (greet_user prompt from fastmcp_quickstart) + if prompts.prompts: + prompt = await session.get_prompt("greet_user", arguments={"name": "Alice", "style": "friendly"}) + print(f"Prompt result: {prompt.messages[0].content}") # List available resources resources = await session.list_resources() + print(f"Available resources: {[r.uri for r in resources.resources]}") # List available tools tools = await session.list_tools() + print(f"Available tools: {[t.name for t in tools.tools]}") - # Read a resource - content, mime_type = await session.read_resource("file://some/path") + # Read a resource (greeting resource from fastmcp_quickstart) + resource_content = await session.read_resource(AnyUrl("greeting://World")) + content_block = resource_content.contents[0] + if isinstance(content_block, types.TextContent): + print(f"Resource content: {content_block.text}") - # Call a tool - result = await session.call_tool("tool-name", arguments={"arg1": "value"}) + # Call a tool (add tool from fastmcp_quickstart) + result = await session.call_tool("add", arguments={"a": 5, "b": 3}) + result_unstructured = result.content[0] + if isinstance(result_unstructured, types.TextContent): + print(f"Tool result: {result_unstructured.text}") + result_structured = result.structuredContent + print(f"Structured tool result: {result_structured}") -if __name__ == "__main__": - import asyncio - +def main(): + """Entry point for the client script.""" asyncio.run(run()) + + +if __name__ == "__main__": + main() ``` +_Full example: [examples/snippets/clients/stdio_client.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/stdio_client.py)_ + + Clients can also connect using [Streamable HTTP transport](https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#streamable-http): ```python diff --git a/examples/snippets/clients/__init__.py b/examples/snippets/clients/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/examples/snippets/clients/completion_client.py b/examples/snippets/clients/completion_client.py new file mode 100644 index 000000000..d8745ea1e --- /dev/null +++ b/examples/snippets/clients/completion_client.py @@ -0,0 +1,80 @@ +"""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 +""" + +import asyncio +import os + +from mcp import ClientSession, StdioServerParameters +from mcp.client.stdio import stdio_client +from mcp.types import PromptReference, ResourceTemplateReference + +# Create server parameters for stdio connection +server_params = StdioServerParameters( + command="uv", # Using uv to run the server + args=["run", "server", "completion", "stdio"], # Server with completion support + env={"UV_INDEX": os.environ.get("UV_INDEX", "")}, +) + + +async def run(): + """Run the completion client example.""" + async with stdio_client(server_params) as (read, write): + async with ClientSession(read, write) as session: + # Initialize the connection + await session.initialize() + + # List available resource templates + templates = await session.list_resource_templates() + print("Available resource templates:") + for template in templates.resourceTemplates: + print(f" - {template.uriTemplate}") + + # List available prompts + prompts = await session.list_prompts() + print("\nAvailable prompts:") + for prompt in prompts.prompts: + print(f" - {prompt.name}") + + # Complete resource template arguments + if templates.resourceTemplates: + template = templates.resourceTemplates[0] + print(f"\nCompleting arguments for resource template: {template.uriTemplate}") + + # Complete without context + result = await session.complete( + ref=ResourceTemplateReference(type="ref/resource", uri=template.uriTemplate), + argument={"name": "owner", "value": "model"}, + ) + print(f"Completions for 'owner' starting with 'model': {result.completion.values}") + + # Complete with context - repo suggestions based on owner + result = await session.complete( + ref=ResourceTemplateReference(type="ref/resource", uri=template.uriTemplate), + argument={"name": "repo", "value": ""}, + context_arguments={"owner": "modelcontextprotocol"}, + ) + print(f"Completions for 'repo' with owner='modelcontextprotocol': {result.completion.values}") + + # Complete prompt arguments + if prompts.prompts: + prompt_name = prompts.prompts[0].name + print(f"\nCompleting arguments for prompt: {prompt_name}") + + result = await session.complete( + ref=PromptReference(type="ref/prompt", name=prompt_name), + argument={"name": "style", "value": ""}, + ) + print(f"Completions for 'style' argument: {result.completion.values}") + + +def main(): + """Entry point for the completion client.""" + asyncio.run(run()) + + +if __name__ == "__main__": + main() diff --git a/examples/snippets/clients/stdio_client.py b/examples/snippets/clients/stdio_client.py new file mode 100644 index 000000000..22ad933ad --- /dev/null +++ b/examples/snippets/clients/stdio_client.py @@ -0,0 +1,85 @@ +"""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 +""" + +import asyncio +import os + +from pydantic import AnyUrl + +from mcp import ClientSession, StdioServerParameters, types +from mcp.client.stdio import stdio_client +from mcp.shared.context import RequestContext + +# Create server parameters for stdio connection +server_params = StdioServerParameters( + command="uv", # Using uv to run the server + args=["run", "server", "fastmcp_quickstart", "stdio"], # We're already in snippets dir + env={"UV_INDEX": os.environ.get("UV_INDEX", "")}, +) + + +# Optional: create a sampling callback +async def handle_sampling_message( + context: RequestContext, params: types.CreateMessageRequestParams +) -> types.CreateMessageResult: + print(f"Sampling request: {params.messages}") + 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() + print(f"Available prompts: {[p.name for p in prompts.prompts]}") + + # Get a prompt (greet_user prompt from fastmcp_quickstart) + if prompts.prompts: + prompt = await session.get_prompt("greet_user", arguments={"name": "Alice", "style": "friendly"}) + print(f"Prompt result: {prompt.messages[0].content}") + + # List available resources + resources = await session.list_resources() + print(f"Available resources: {[r.uri for r in resources.resources]}") + + # List available tools + tools = await session.list_tools() + print(f"Available tools: {[t.name for t in tools.tools]}") + + # Read a resource (greeting resource from fastmcp_quickstart) + resource_content = await session.read_resource(AnyUrl("greeting://World")) + content_block = resource_content.contents[0] + if isinstance(content_block, types.TextContent): + print(f"Resource content: {content_block.text}") + + # Call a tool (add tool from fastmcp_quickstart) + result = await session.call_tool("add", arguments={"a": 5, "b": 3}) + result_unstructured = result.content[0] + if isinstance(result_unstructured, types.TextContent): + print(f"Tool result: {result_unstructured.text}") + result_structured = result.structuredContent + print(f"Structured tool result: {result_structured}") + + +def main(): + """Entry point for the client script.""" + asyncio.run(run()) + + +if __name__ == "__main__": + main() diff --git a/examples/snippets/pyproject.toml b/examples/snippets/pyproject.toml index 832f6495b..e0ca9ba6b 100644 --- a/examples/snippets/pyproject.toml +++ b/examples/snippets/pyproject.toml @@ -12,7 +12,9 @@ requires = ["setuptools", "wheel"] build-backend = "setuptools.build_meta" [tool.setuptools] -packages = ["servers"] +packages = ["servers", "clients"] [project.scripts] -server = "servers:run_server" \ No newline at end of file +server = "servers:run_server" +client = "clients.stdio_client:main" +completion-client = "clients.completion_client:main" diff --git a/examples/snippets/servers/__init__.py b/examples/snippets/servers/__init__.py index e3b778420..b9865e822 100644 --- a/examples/snippets/servers/__init__.py +++ b/examples/snippets/servers/__init__.py @@ -21,7 +21,8 @@ def run_server(): if len(sys.argv) < 2: print("Usage: server [transport]") print("Available servers: basic_tool, basic_resource, basic_prompt, tool_progress,") - print(" sampling, elicitation, completion, notifications") + print(" sampling, elicitation, completion, notifications,") + print(" fastmcp_quickstart, structured_output, images") print("Available transports: stdio (default), sse, streamable-http") sys.exit(1) diff --git a/examples/snippets/servers/fastmcp_quickstart.py b/examples/snippets/servers/fastmcp_quickstart.py new file mode 100644 index 000000000..d7aef8c61 --- /dev/null +++ b/examples/snippets/servers/fastmcp_quickstart.py @@ -0,0 +1,38 @@ +""" +FastMCP quickstart example. + +cd to the `examples/snippets/clients` directory and run: + uv run server fastmcp_quickstart stdio +""" + +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}!" + + +# Add a prompt +@mcp.prompt() +def greet_user(name: str, style: str = "friendly") -> str: + """Generate a greeting prompt""" + styles = { + "friendly": "Please write a warm, friendly greeting", + "formal": "Please write a formal, professional greeting", + "casual": "Please write a casual, relaxed greeting", + } + + return f"{styles.get(style, styles['friendly'])} for someone named {name}." diff --git a/examples/snippets/servers/images.py b/examples/snippets/servers/images.py new file mode 100644 index 000000000..9e0262c85 --- /dev/null +++ b/examples/snippets/servers/images.py @@ -0,0 +1,15 @@ +"""Example showing image handling with FastMCP.""" + +from PIL import Image as PILImage + +from mcp.server.fastmcp import FastMCP, Image + +mcp = FastMCP("Image Example") + + +@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") diff --git a/examples/snippets/servers/lifespan_example.py b/examples/snippets/servers/lifespan_example.py new file mode 100644 index 000000000..37d04b597 --- /dev/null +++ b/examples/snippets/servers/lifespan_example.py @@ -0,0 +1,56 @@ +"""Example showing lifespan support for startup/shutdown with strong typing.""" + +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from dataclasses import dataclass + +from mcp.server.fastmcp import Context, FastMCP + + +# Mock database class for example +class Database: + """Mock database class for example.""" + + @classmethod + async def connect(cls) -> "Database": + """Connect to database.""" + return cls() + + async def disconnect(self) -> None: + """Disconnect from database.""" + pass + + def query(self) -> str: + """Execute a query.""" + return "Query result" + + +@dataclass +class AppContext: + """Application context with typed dependencies.""" + + 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() diff --git a/examples/snippets/servers/structured_output.py b/examples/snippets/servers/structured_output.py new file mode 100644 index 000000000..4cce69d8c --- /dev/null +++ b/examples/snippets/servers/structured_output.py @@ -0,0 +1,29 @@ +"""Example showing structured output with tools.""" + +from pydantic import BaseModel, Field + +from mcp.server.fastmcp import FastMCP + +mcp = FastMCP("Structured Output Example") + + +# Using Pydantic models for rich structured data +class WeatherData(BaseModel): + """Weather information structure.""" + + temperature: float = Field(description="Temperature in Celsius") + humidity: float = Field(description="Humidity percentage") + condition: str + wind_speed: float + + +@mcp.tool() +def get_weather(city: str) -> WeatherData: + """Get weather for a city - returns structured data.""" + # Simulated weather data + return WeatherData( + temperature=72.5, + humidity=45.0, + condition="sunny", + wind_speed=5.2, + ) diff --git a/pyproject.toml b/pyproject.toml index 878c8d184..474c58f6e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -88,7 +88,7 @@ Issues = "https://github.com/modelcontextprotocol/python-sdk/issues" packages = ["src/mcp"] [tool.pyright] -include = ["src/mcp", "tests", "examples/servers"] +include = ["src/mcp", "tests", "examples/servers", "examples/snippets"] venvPath = "." venv = ".venv" strict = ["src/mcp/**/*.py"] diff --git a/tests/server/fastmcp/test_integration.py b/tests/server/fastmcp/test_integration.py index a1620ca17..377e4923b 100644 --- a/tests/server/fastmcp/test_integration.py +++ b/tests/server/fastmcp/test_integration.py @@ -21,8 +21,10 @@ basic_tool, completion, elicitation, + fastmcp_quickstart, notifications, sampling, + structured_output, tool_progress, ) from mcp.client.session import ClientSession @@ -100,6 +102,10 @@ def run_server_with_transport(module_name: str, port: int, transport: str) -> No mcp = completion.mcp elif module_name == "notifications": mcp = notifications.mcp + elif module_name == "fastmcp_quickstart": + mcp = fastmcp_quickstart.mcp + elif module_name == "structured_output": + mcp = structured_output.mcp else: raise ImportError(f"Unknown module: {module_name}") @@ -590,3 +596,77 @@ async def test_completion(server_transport: str, server_url: str) -> None: assert completion_result.completion is not None assert "python" in completion_result.completion.values assert all(lang.startswith("py") for lang in completion_result.completion.values) + + +# Test FastMCP quickstart example +@pytest.mark.anyio +@pytest.mark.parametrize( + "server_transport", + [ + ("fastmcp_quickstart", "sse"), + ("fastmcp_quickstart", "streamable-http"), + ], + indirect=True, +) +async def test_fastmcp_quickstart(server_transport: str, server_url: str) -> None: + """Test FastMCP quickstart example.""" + transport = server_transport + client_cm = create_client_for_transport(transport, server_url) + + async with client_cm as client_streams: + read_stream, write_stream = unpack_streams(client_streams) + async with ClientSession(read_stream, write_stream) as session: + # Test initialization + result = await session.initialize() + assert isinstance(result, InitializeResult) + assert result.serverInfo.name == "Demo" + + # Test add tool + tool_result = await session.call_tool("add", {"a": 10, "b": 20}) + assert len(tool_result.content) == 1 + assert isinstance(tool_result.content[0], TextContent) + assert tool_result.content[0].text == "30" + + # Test greeting resource directly + from pydantic import AnyUrl + + resource_result = await session.read_resource(AnyUrl("greeting://Alice")) + assert len(resource_result.contents) == 1 + assert isinstance(resource_result.contents[0], TextResourceContents) + assert resource_result.contents[0].text == "Hello, Alice!" + + +# Test structured output example +@pytest.mark.anyio +@pytest.mark.parametrize( + "server_transport", + [ + ("structured_output", "sse"), + ("structured_output", "streamable-http"), + ], + indirect=True, +) +async def test_structured_output(server_transport: str, server_url: str) -> None: + """Test structured output functionality.""" + transport = server_transport + client_cm = create_client_for_transport(transport, server_url) + + async with client_cm as client_streams: + read_stream, write_stream = unpack_streams(client_streams) + async with ClientSession(read_stream, write_stream) as session: + # Test initialization + result = await session.initialize() + assert isinstance(result, InitializeResult) + assert result.serverInfo.name == "Structured Output Example" + + # Test get_weather tool + weather_result = await session.call_tool("get_weather", {"city": "New York"}) + assert len(weather_result.content) == 1 + assert isinstance(weather_result.content[0], TextContent) + + # Check that the result contains expected weather data + result_text = weather_result.content[0].text + assert "72.5" in result_text # temperature + assert "sunny" in result_text # condition + assert "45" in result_text # humidity + assert "5.2" in result_text # wind_speed