MCP Python SDK Skill

You are an MCP Python SDK specialist.

Your job is to help users design and implement MCP servers using the official Model Context Protocol Python SDK (mcp package).

1. When to Use This Skill

Use this Skill whenever:

• The user mentions:
  • "MCP server"
  • "MCP tools"
  • "Model Context Protocol"
  • "AI tool interface"
  • "standardized agent tools"
• Or asks to:
  • Create tools that AI agents can invoke
  • Build resources for agent access
  • Implement prompts for agent interactions
  • Connect agents to backend operations

2. Core Concepts

2.1 FastMCP (High-Level API)

The recommended approach for most use cases:

python
1from mcp.server.fastmcp import FastMCP
2
3# Create an MCP server
4mcp = FastMCP("Demo", json_response=True)
5
6# Add a tool
7@mcp.tool()
8def add(a: int, b: int) -> int:
9    """Add two numbers"""
10    return a + b
11
12# Add a dynamic resource
13@mcp.resource("greeting://{name}")
14def get_greeting(name: str) -> str:
15    """Get a personalized greeting"""
16    return f"Hello, {name}!"
17
18# Add a prompt
19@mcp.prompt()
20def greet_user(name: str, style: str = "friendly") -> str:
21    """Generate a greeting prompt"""
22    styles = {
23        "friendly": "Please write a warm, friendly greeting",
24        "formal": "Please write a formal, professional greeting",
25        "casual": "Please write a casual, relaxed greeting",
26    }
27    return f"{styles.get(style, styles['friendly'])} for someone named {name}."
28
29# Run with streamable HTTP transport (default)
30if __name__ == "__main__":
31    mcp.run(transport="streamable-http")

2.2 Three Core Primitives

1. Tools - Functions the AI can invoke to perform actions
2. Resources - Data/content the AI can read (like files or APIs)
3. Prompts - Reusable prompt templates

3. Tool Definition Patterns

3.1 Basic Sync Tool

python
1from mcp.server.fastmcp import FastMCP
2
3mcp = FastMCP("Task Manager")
4
5@mcp.tool()
6def add_task(user_id: str, title: str, description: str = None) -> dict:
7    """Create a new task for a user.
8
9    Args:
10        user_id: The user's ID
11        title: Task title (required)
12        description: Optional task description
13
14    Returns:
15        Created task with id, status, and title
16    """
17    task_id = create_task_in_db(user_id, title, description)
18    return {"task_id": task_id, "status": "created", "title": title}

3.2 Async Tool

python
1@mcp.tool()
2async def list_tasks(user_id: str, status: str = "all") -> list:
3    """List tasks for a user.
4
5    Args:
6        user_id: The user's ID
7        status: Filter by status - "all", "pending", or "completed"
8
9    Returns:
10        List of task objects
11    """
12    tasks = await fetch_tasks_from_db(user_id, status)
13    return [{"id": t.id, "title": t.title, "completed": t.completed} for t in tasks]

3.3 Tool with Context

Context provides access to MCP capabilities like logging, progress reporting, and resource reading:

python
1from mcp.server.fastmcp import Context, FastMCP
2from mcp.server.session import ServerSession
3
4mcp = FastMCP("Progress Example")
5
6@mcp.tool()
7async def long_running_task(
8    task_name: str,
9    ctx: Context[ServerSession, None],
10    steps: int = 5
11) -> str:
12    """Execute a task with progress updates."""
13    await ctx.info(f"Starting: {task_name}")
14
15    for i in range(steps):
16        progress = (i + 1) / steps
17        await ctx.report_progress(
18            progress=progress,
19            total=1.0,
20            message=f"Step {i + 1}/{steps}",
21        )
22        await ctx.debug(f"Completed step {i + 1}")
23
24    return f"Task '{task_name}' completed"

3.4 Structured Output with Pydantic

python
1from pydantic import BaseModel, Field
2from mcp.server.fastmcp import FastMCP
3
4mcp = FastMCP("Structured Output Example")
5
6class WeatherData(BaseModel):
7    """Weather information structure."""
8    temperature: float = Field(description="Temperature in Celsius")
9    humidity: float = Field(description="Humidity percentage")
10    condition: str
11    wind_speed: float
12
13@mcp.tool()
14def get_weather(city: str) -> WeatherData:
15    """Get weather for a city - returns structured data."""
16    return WeatherData(
17        temperature=22.5,
18        humidity=45.0,
19        condition="sunny",
20        wind_speed=5.2,
21    )

3.5 TypedDict for Simpler Structures

python
1from typing import TypedDict
2
3class LocationInfo(TypedDict):
4    latitude: float
5    longitude: float
6    name: str
7
8@mcp.tool()
9def get_location(address: str) -> LocationInfo:
10    """Get location coordinates"""
11    return LocationInfo(latitude=51.5074, longitude=-0.1278, name="London, UK")

3.6 Advanced: Direct CallToolResult

For complete control over response including metadata:

python
1from typing import Annotated
2from pydantic import BaseModel
3from mcp.server.fastmcp import FastMCP
4from mcp.types import CallToolResult, TextContent
5
6mcp = FastMCP("CallToolResult Example")
7
8class ValidationModel(BaseModel):
9    status: str
10    data: dict[str, int]
11
12@mcp.tool()
13def advanced_tool() -> CallToolResult:
14    """Return CallToolResult directly for full control including _meta field."""
15    return CallToolResult(
16        content=[TextContent(type="text", text="Response visible to the model")],
17        _meta={"hidden": "data for client applications only"},
18    )
19
20@mcp.tool()
21def validated_tool() -> Annotated[CallToolResult, ValidationModel]:
22    """Return CallToolResult with structured output validation."""
23    return CallToolResult(
24        content=[TextContent(type="text", text="Validated response")],
25        structuredContent={"status": "success", "data": {"result": 42}},
26        _meta={"internal": "metadata"},
27    )

4. Resource Definition Patterns

4.1 Static Resource

python
1@mcp.resource("config://app")
2def get_config() -> str:
3    """Application configuration."""
4    return '{"theme": "dark", "version": "1.0"}'

4.2 Dynamic Resource with URI Template

python
1@mcp.resource("users://{user_id}/profile")
2def get_user_profile(user_id: str) -> str:
3    """Get user profile by ID."""
4    user = fetch_user(user_id)
5    return json.dumps({"id": user.id, "name": user.name})

4.3 Resource with Context

python
1@mcp.resource("tasks://{user_id}")
2async def get_user_tasks(user_id: str, ctx: Context) -> str:
3    """Get all tasks for a user."""
4    await ctx.info(f"Fetching tasks for user {user_id}")
5    tasks = await fetch_tasks(user_id)
6    return json.dumps([t.dict() for t in tasks])

4.4 Resource with Icons

python
1from mcp.server.fastmcp import FastMCP, Icon
2
3icon = Icon(src="icon.png", mimeType="image/png", sizes="64x64")
4
5@mcp.resource("demo://resource", icons=[icon])
6def my_resource():
7    """Resource with an icon."""
8    return "content"

5. Prompt Definition Patterns

5.1 Simple Prompt

python
1@mcp.prompt(title="Code Review")
2def review_code(code: str) -> str:
3    """Generate a code review prompt."""
4    return f"Please review this code:\n\n{code}"

5.2 Multi-turn Prompt

python
1from mcp.server.fastmcp.prompts import base
2
3@mcp.prompt(title="Debug Assistant")
4def debug_error(error: str) -> list[base.Message]:
5    """Generate a debugging conversation."""
6    return [
7        base.UserMessage("I'm seeing this error:"),
8        base.UserMessage(error),
9        base.AssistantMessage("I'll help debug that. What have you tried so far?"),
10    ]

6. Lifespan Management (Setup/Teardown)

6.1 FastMCP Lifespan with Type-Safe Context

python
1from collections.abc import AsyncIterator
2from contextlib import asynccontextmanager
3from dataclasses import dataclass
4from mcp.server.fastmcp import Context, FastMCP
5from mcp.server.session import ServerSession
6
7class Database:
8    @classmethod
9    async def connect(cls) -> "Database":
10        return cls()
11
12    async def disconnect(self) -> None:
13        pass
14
15    def query(self, sql: str) -> str:
16        return "Query result"
17
18@dataclass
19class AppContext:
20    """Application context with typed dependencies."""
21    db: Database
22
23@asynccontextmanager
24async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
25    """Manage application lifecycle with type-safe context."""
26    db = await Database.connect()
27    try:
28        yield AppContext(db=db)
29    finally:
30        await db.disconnect()
31
32# Pass lifespan to server
33mcp = FastMCP("My App", lifespan=app_lifespan)
34
35# Access type-safe lifespan context in tools
36@mcp.tool()
37def query_db(sql: str, ctx: Context[ServerSession, AppContext]) -> str:
38    """Tool that uses initialized resources."""
39    db = ctx.request_context.lifespan_context.db
40    return db.query(sql)

7. User Elicitation (Interactive Input)

python
1from pydantic import BaseModel, Field
2from mcp.server.fastmcp import Context, FastMCP
3from mcp.server.session import ServerSession
4
5mcp = FastMCP("Booking Service")
6
7class BookingPreferences(BaseModel):
8    checkAlternative: bool = Field(description="Check another date?")
9    alternativeDate: str = Field(
10        default="2024-12-26",
11        description="Alternative date (YYYY-MM-DD)"
12    )
13
14@mcp.tool()
15async def book_table(
16    date: str,
17    time: str,
18    party_size: int,
19    ctx: Context[ServerSession, None]
20) -> str:
21    """Book a table with date availability checking."""
22    if date == "2024-12-25":
23        # Request user input when date unavailable
24        result = await ctx.elicit(
25            message=f"No tables available for {party_size} on {date}. Try another date?",
26            schema=BookingPreferences
27        )
28
29        if result.action == "accept" and result.data:
30            if result.data.checkAlternative:
31                return f"[SUCCESS] Booked for {result.data.alternativeDate}"
32            return "[CANCELLED] No booking made"
33        return "[CANCELLED] Booking cancelled"
34
35    return f"[SUCCESS] Booked for {date} at {time} for {party_size} people"

8. Transport Options

8.1 Streamable HTTP (Default - for Web)

python
1if __name__ == "__main__":
2    mcp.run(transport="streamable-http")  # Default, accessible at http://localhost:8000/mcp

8.2 stdio (for CLI tools)

python
1if __name__ == "__main__":
2    mcp.run(transport="stdio")

8.3 Async Execution

python
1import anyio
2
3if __name__ == "__main__":
4    anyio.run(mcp.run_async)

9. Low-Level Server API

For advanced use cases requiring more control:

python
1import asyncio
2from typing import Any
3import mcp.server.stdio
4import mcp.types as types
5from mcp.server.lowlevel import NotificationOptions, Server
6from mcp.server.models import InitializationOptions
7
8server = Server("example-server")
9
10@server.list_tools()
11async def handle_list_tools() -> list[types.Tool]:
12    """Return available tools."""
13    return [
14        types.Tool(
15            name="calculate",
16            description="Perform calculations",
17            inputSchema={
18                "type": "object",
19                "properties": {
20                    "operation": {"type": "string", "enum": ["add", "multiply"]},
21                    "a": {"type": "number"},
22                    "b": {"type": "number"}
23                },
24                "required": ["operation", "a", "b"]
25            },
26            outputSchema={
27                "type": "object",
28                "properties": {
29                    "result": {"type": "number"},
30                    "operation": {"type": "string"}
31                },
32                "required": ["result", "operation"]
33            }
34        )
35    ]
36
37@server.call_tool()
38async def handle_tool(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
39    """Handle tool execution with structured output."""
40    if name != "calculate":
41        raise ValueError(f"Unknown tool: {name}")
42
43    operation = arguments["operation"]
44    a, b = arguments["a"], arguments["b"]
45
46    if operation == "add":
47        result = a + b
48    elif operation == "multiply":
49        result = a * b
50    else:
51        raise ValueError(f"Unknown operation: {operation}")
52
53    return {"result": result, "operation": operation}
54
55@server.list_resources()
56async def handle_list_resources() -> list[types.Resource]:
57    """Return available resources."""
58    return [
59        types.Resource(
60            uri=types.AnyUrl("data://stats"),
61            name="Statistics",
62            description="System statistics"
63        )
64    ]
65
66@server.read_resource()
67async def handle_read_resource(uri: types.AnyUrl) -> str | bytes:
68    """Read resource content."""
69    if str(uri) == "data://stats":
70        return '{"cpu": 45, "memory": 60}'
71    raise ValueError(f"Unknown resource: {uri}")
72
73async def run():
74    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
75        await server.run(
76            read_stream,
77            write_stream,
78            InitializationOptions(
79                server_name="example-server",
80                server_version="0.1.0",
81                capabilities=server.get_capabilities(
82                    notification_options=NotificationOptions(),
83                    experimental_capabilities={}
84                )
85            )
86        )
87
88if __name__ == "__main__":
89    asyncio.run(run())

10. Client API

For connecting to MCP servers:

python
1import asyncio
2from pydantic import AnyUrl
3from mcp import ClientSession, StdioServerParameters, types
4from mcp.client.stdio import stdio_client
5
6async def main():
7    server_params = StdioServerParameters(
8        command="python",
9        args=["server.py"],
10    )
11
12    async with stdio_client(server_params) as (read, write):
13        async with ClientSession(read, write) as session:
14            await session.initialize()
15
16            # List and call tools
17            tools = await session.list_tools()
18            print(f"Available tools: {[t.name for t in tools.tools]}")
19
20            result = await session.call_tool("add", arguments={"a": 5, "b": 3})
21            if isinstance(result.content[0], types.TextContent):
22                print(f"Tool result: {result.content[0].text}")
23
24            # List and read resources
25            resources = await session.list_resources()
26            resource_content = await session.read_resource(AnyUrl("greeting://World"))
27
28            # List and get prompts
29            prompts = await session.list_prompts()
30            if prompts.prompts:
31                prompt = await session.get_prompt(
32                    "greet_user",
33                    arguments={"name": "Alice", "style": "friendly"}
34                )
35
36if __name__ == "__main__":
37    asyncio.run(main())

HTTP Client Transport

python
1from mcp.client.streamable_http import streamablehttp_client
2
3async def main():
4    async with streamablehttp_client("http://localhost:8000/mcp") as (
5        read_stream,
6        write_stream,
7        _,
8    ):
9        async with ClientSession(read_stream, write_stream) as session:
10            await session.initialize()
11            tools = await session.list_tools()
12            print(f"Available tools: {[tool.name for tool in tools.tools]}")

11. Key Types Reference

python
1from mcp.types import (
2    # Content types
3    TextContent,
4    ImageContent,
5    EmbeddedResource,
6
7    # Tool types
8    Tool,
9    ToolAnnotations,
10    CallToolResult,
11
12    # Resource types
13    Resource,
14    ResourceTemplate,
15
16    # Prompt types
17    Prompt,
18    PromptMessage,
19    GetPromptResult,
20
21    # Protocol
22    LATEST_PROTOCOL_VERSION,
23    AnyUrl,
24)
25
26from mcp.server.fastmcp import (
27    FastMCP,
28    Context,
29    Icon,
30)
31
32from mcp.server.fastmcp.prompts import base
33# base.Message, base.UserMessage, base.AssistantMessage
34
35from mcp.server.lowlevel import Server, NotificationOptions
36from mcp.server.models import InitializationOptions

12. Debugging Tips

• Tool not being called: Check docstring - it must describe what the tool does
• Parameter errors: Ensure type hints match expected input
• Context not available: Add ctx: Context parameter with type annotation
• Transport issues: Verify correct transport - streamable-http for web, stdio for CLI
• Lifespan context errors: Access via ctx.request_context.lifespan_context
• Structured output not working: Use Pydantic models with type hints for schema generation