modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

README - replace code snippets with examples (#1136)

Browse source
This commit is contained in:
Inna Harper 2025-07-14 13:57:47 +01:00 • committed by GitHub
commit 9bf5160eea
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 585 additions and 169 deletions
Download patch file Download diff file Expand all files Collapse all files

372
README.md
View file

@ -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:
<!-- snippet-source examples/snippets/servers/fastmcp_quickstart.py -->
```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)_
<!-- /snippet-source -->
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:
<!-- snippet-source examples/snippets/servers/lifespan_example.py -->
```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."""
# Specify dependencies for deployment and development
mcp = FastMCP("My App", dependencies=["pandas", "numpy"])
@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"""
"""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)_
<!-- /snippet-source -->
### 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.
<!-- snippet-source examples/snippets/servers/structured_output.py -->
```python
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field
from typing import TypedDict
"""Example showing structured output with tools."""
mcp = FastMCP("Weather Service")
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
@ -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)_
<!-- /snippet-source -->
```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:
<!-- snippet-source examples/snippets/servers/images.py -->
```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)_
<!-- /snippet-source -->
### 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:
<!-- snippet-source examples/snippets/clients/completion_client.py -->
```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"},
)
import asyncio
import os
# 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"},
)
```
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.types import PromptReference, ResourceTemplateReference
Server implementation:
<!-- snippet-source examples/snippets/servers/completion.py -->
```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}")
# 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}")
@mcp.resource("github://repos/{owner}/{repo}")
def github_repo(owner: str, repo: str) -> str:
"""GitHub repository resource."""
return f"Repository: {owner}/{repo}"
def main():
"""Entry point for the completion client."""
asyncio.run(run())
@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}"
@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,
)
# 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)_
<!-- /snippet-source -->
### 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):
<!-- snippet-source examples/snippets/clients/stdio_client.py -->
```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}")
def main():
"""Entry point for the client script."""
asyncio.run(run())
if __name__ == "__main__":
import asyncio
asyncio.run(run())
main()
```
_Full example: [examples/snippets/clients/stdio_client.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/stdio_client.py)_
<!-- /snippet-source -->
Clients can also connect using [Streamable HTTP transport](https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#streamable-http):
```python

0
examples/snippets/clients/__init__.py Normal file
View file

80
examples/snippets/clients/completion_client.py Normal file
View file

@ -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()

85
examples/snippets/clients/stdio_client.py Normal file
View file

@ -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()

6
examples/snippets/pyproject.toml
View file

@ -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"
server = "servers:run_server"
client = "clients.stdio_client:main"
completion-client = "clients.completion_client:main"

3
examples/snippets/servers/__init__.py
View file

@ -21,7 +21,8 @@ def run_server():
if len(sys.argv) < 2:
print("Usage: server <server-name> [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)

38
examples/snippets/servers/fastmcp_quickstart.py Normal file
View file

@ -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}."

15
examples/snippets/servers/images.py Normal file
View file

@ -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")

56
examples/snippets/servers/lifespan_example.py Normal file
View file

@ -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()

29
examples/snippets/servers/structured_output.py Normal file
View file

@ -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,
)

2
pyproject.toml
View file

@ -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"]

80
tests/server/fastmcp/test_integration.py
View file

@ -21,8 +21,10 @@ from examples.snippets.servers import (
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