mirror of
https://github.com/github/awesome-copilot.git
synced 2026-10-13 15:24:25 +00:00
248 lines
9.7 KiB
Markdown
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!"}
|
|
```
|