Files
awesome-copilot/instructions/python-mcp-server.instructions.md
T

248 lines
9.7 KiB
Markdown

---
description: 'Instructions for building Model Context Protocol (MCP) servers using the Python SDK v2'
applyTo: '**/*.py, **/pyproject.toml, **/requirements.txt'
---
# Python MCP Server Development
These instructions apply to the [MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk) v2 and the MCP specification revision [2026-07-28](https://modelcontextprotocol.io/specification/2026-07-28).
## Instructions
- Use **uv** for project management: `uv init --app --no-package mcp-server-demo` and `uv add "mcp[cli]>=2,<3"`.
- Import the server class from `mcp.server`: `from mcp.server import MCPServer`.
- Do not use `FastMCP` or `mcp.server.fastmcp`. SDK v2 removed this import path.
- Import `Context`, `Image`, `Audio`, `Resolve`, `Elicit`, and the prompt message classes from `mcp.server.mcpserver`.
- Use the `@mcp.tool()`, `@mcp.resource()`, and `@mcp.prompt()` decorators for registration.
- Add type hints to all parameters. The SDK makes the schemas from the type hints and validates the arguments.
- Use Pydantic models, TypedDicts, or dataclasses for structured output.
- A tool returns structured output when its return type is compatible.
- For stdio transport, use `mcp.run()` or `mcp.run(transport="stdio")`.
- For HTTP servers, use `mcp.run(transport="streamable-http")`, or mount `mcp.streamable_http_app()` in Starlette or FastAPI.
- Give transport options (`host`, `port`, `json_response`, `stateless_http`, `transport_security`) to `run()`, not to `MCPServer(...)`.
- When you mount the server, give the same options to `streamable_http_app()`. This method has no `port` option.
- Give the server name as the first `MCPServer` argument. Give all other constructor arguments as keyword arguments.
- Set `version` on `MCPServer`. If you do not set it, the server reports an empty version.
- The SDK v2 types use snake_case fields, for example `read_only_hint`, `structured_content`, and `is_error`.
- Add a `ctx: Context` parameter to a tool, resource, or prompt to get the request context.
- Report progress with `await ctx.report_progress(progress, total, message)`.
- Get user input with a `Resolve(fn)` parameter. Return `Elicit(message, Model)` from `fn`. This works with clients of all protocol revisions.
- Do not use `ctx.elicit()`. It fails on a 2026-07-28 connection.
- Log with the standard `logging` module. Protocol logging (`ctx.info()` and similar methods) is deprecated.
- Do not use sampling (`ctx.session.create_message()`). It is deprecated. Call the LLM provider API directly.
- Do not use roots (`ctx.session.list_roots()`). It is deprecated. Get paths from tool parameters or the server configuration.
- Configure icons with `Icon(src="https://...", mime_type="image/png")` from `mcp.types`. Server, tools, resources, and prompts accept `icons=[...]`.
- Use the `Image` class to return images: `return Image(data=png_bytes, format="png")` or `return Image(path=file_path)`.
- Define resource templates with RFC 6570 URI templates: `@mcp.resource("greeting://{name}")`.
- Add completion support with the `@mcp.completion()` decorator.
- Use a lifespan context manager for startup and shutdown of shared resources. For stdio and HTTP, the lifespan runs one time for the server. Each in-memory `Client(mcp)` runs it again.
- Get the lifespan object in tools from `ctx.request_context.lifespan_context`. Use `Context[AppContext]` for a typed result.
- Raise `ToolError` (from `mcp.server.mcpserver.exceptions`) for an error that the model must read.
- Do not raise `MCPError` for a tool failure. The client gets a protocol error, not a tool result with `is_error`. Many hosts do not show this error to the model.
- Send list change notifications with `await ctx.notify_tools_changed()`. Do not use `ctx.session.send_tool_list_changed()`. A 2026-07-28 connection drops it.
- On the 2026-07-28 revision, Streamable HTTP requests have no session. `stateless_http=True` changes only how the server serves clients of earlier revisions.
- `json_response=True` sends one JSON body for each request. In this mode, the server cannot send progress notifications during the request.
- Test servers with `uv run mcp dev server.py` (MCP Inspector) or `uv run mcp install server.py` (Claude Desktop).
- Write tests with the in-memory client: `async with Client(mcp) as client:` (`from mcp import Client`).
- Mount more than one server in Starlette with different paths: `Mount("/path", app=mcp.streamable_http_app())`. The lifespan of the host app must enter `mcp.session_manager.run()` for each server.
- For browser clients, configure CORS. Allow the `Mcp-*` request headers. Expose the `Mcp-Session-Id` response header.
- If you do not set `transport_security`, the SDK checks the `Host` and `Origin` headers only when the host is `127.0.0.1`, `localhost`, or `::1`. For all other hosts, set `transport_security=TransportSecuritySettings(allowed_hosts=[...], allowed_origins=[...])` (from `mcp.server.transport_security`).
- For more than one worker, set `request_state_security=RequestStateSecurity(keys=[...])` on `MCPServer`.
- Use the low-level `Server` class (from `mcp.server`) only when `MCPServer` does not give sufficient control.
## Best Practices
- Give each tool function one responsibility.
- Write clear docstrings. They become the tool descriptions.
- Use descriptive parameter names with type hints.
- Validate inputs with Pydantic `Field` descriptions.
- Use async functions for I/O-bound operations. The SDK runs sync functions on a worker thread.
- Log to stderr. Do not use `print()` in a stdio server, because stdout is the protocol channel.
- Use environment variables for configuration.
- Be careful when a tool gives access to the file system or the network.
## Common Patterns
### Basic Server Setup (stdio)
```python
from mcp.server import MCPServer
mcp = MCPServer("My Server", version="0.1.0")
@mcp.tool()
def calculate(a: int, b: int, op: str) -> int:
"""Perform calculation"""
if op == "add":
return a + b
return a - b
if __name__ == "__main__":
mcp.run() # stdio by default
```
### HTTP Server
```python
from mcp.server import MCPServer
mcp = MCPServer("My HTTP Server", version="0.1.0")
@mcp.tool()
def hello(name: str = "World") -> str:
"""Greet someone"""
return f"Hello, {name}!"
if __name__ == "__main__":
mcp.run(transport="streamable-http", host="127.0.0.1", port=8000)
```
### Tool with Structured Output
```python
from pydantic import BaseModel, Field
class WeatherData(BaseModel):
temperature: float = Field(description="Temperature in Celsius")
condition: str
humidity: float
@mcp.tool()
def get_weather(city: str) -> WeatherData:
"""Get weather for a city"""
return WeatherData(
temperature=22.5,
condition="sunny",
humidity=65.0
)
```
### Dynamic Resource
```python
@mcp.resource("users://{user_id}")
def get_user(user_id: str) -> str:
"""Get user profile data"""
return f"User {user_id} profile data"
```
### Tool with Context
```python
import logging
from mcp.server.mcpserver import Context
logger = logging.getLogger(__name__)
@mcp.tool()
async def process_data(data: str, ctx: Context) -> str:
"""Process data with progress reports"""
logger.info("Processing: %s", data)
await ctx.report_progress(0.5, 1.0, "Halfway done")
return f"Processed: {data}"
```
### Tool with User Input
```python
from typing import Annotated
from pydantic import BaseModel
from mcp.server.mcpserver import AcceptedElicitation, Elicit, ElicitationResult, Resolve
class Quantity(BaseModel):
copies: int
async def ask_quantity() -> Elicit[Quantity]:
"""Ask the user how many copies to reserve."""
return Elicit("How many copies?", Quantity)
@mcp.tool()
async def reserve(
title: str,
quantity: Annotated[ElicitationResult[Quantity], Resolve(ask_quantity)],
) -> str:
"""Reserve copies of a book"""
if isinstance(quantity, AcceptedElicitation):
return f"Reserved {quantity.data.copies} of {title!r}."
return "Nothing reserved."
```
### Lifespan Management
```python
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from mcp.server import MCPServer
from mcp.server.mcpserver import Context
@dataclass
class AppContext:
db: Database # Your database client
@asynccontextmanager
async def app_lifespan(server: MCPServer) -> AsyncIterator[AppContext]:
db = await Database.connect()
try:
yield AppContext(db=db)
finally:
await db.disconnect()
mcp = MCPServer("My App", version="0.1.0", lifespan=app_lifespan)
@mcp.tool()
def query(sql: str, ctx: Context[AppContext]) -> str:
"""Query database"""
db = ctx.request_context.lifespan_context.db
return db.execute(sql)
```
### Prompt with Messages
```python
from mcp.server.mcpserver import AssistantMessage, Message, UserMessage
@mcp.prompt(title="Code Review")
def review_code(code: str) -> list[Message]:
"""Create code review prompt"""
return [
UserMessage("Review this code:"),
UserMessage(code),
AssistantMessage("I'll review the code for you.")
]
```
### Error Handling
```python
from mcp.server.mcpserver.exceptions import ToolError
@mcp.tool()
async def risky_operation(input: str) -> str:
"""Operation that might fail"""
if not input:
raise ToolError("Input must not be empty.")
result = await perform_operation(input) # Your operation
return f"Success: {result}"
```
### In-Memory Test
```python
import pytest
from mcp import Client
from server import mcp # The HTTP Server example
@pytest.mark.anyio
async def test_hello():
async with Client(mcp) as client:
result = await client.call_tool("hello", {"name": "MCP"})
assert result.structured_content == {"result": "Hello, MCP!"}
```