pip install git+https://github.com/thejens/pydantic-mcp-demo.git

from dataclasses import dataclass
from pydantic_ai import RunContext
from pydantic_ai_mcp import MCPServer

@dataclass
class Deps:
    api_key: str

server = MCPServer(deps=Deps(api_key="sk-…"), name="demo")

@server.tool()
async def whoami(ctx: RunContext[Deps]) -> str:
    """Return the current user."""
    return f"Authenticated as {ctx.deps.api_key[:4]}…"

server.run(transport="stdio")

from pydantic_ai import Agent
from pydantic_ai.toolsets import FunctionToolset
from pydantic_ai_mcp import MCPServer

toolset: FunctionToolset[Deps] = FunctionToolset(id="demo")

@toolset.tool()
async def whoami(ctx: RunContext[Deps]) -> str: ...

# Agent path — unchanged
agent = Agent(model="openai:gpt-4o", toolsets=[toolset])

# MCP path — same toolset, deps called fresh per invocation
server = MCPServer(toolsets=[toolset], deps=make_deps)
server.run(transport="stdio")

from pydantic import BaseModel, ConfigDict, Field
from pydantic_ai_mcp import MCPServer

class Deps(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    # persisted across calls ──────────────────────────────────
    user_id: str | None = None
    notes: dict[str, str] = Field(default_factory=dict)
    # rebuilt each call (not stored) ──────────────────────────
    http: httpx.AsyncClient | None = Field(default=None, exclude=True)

async def make_deps(deps: Deps) -> Deps:
    if deps.user_id is None:
        deps.user_id = await auth()        # runs once per session, then cached
    deps.http = httpx.AsyncClient()        # fresh every call
    return deps

server = MCPServer(
    deps=make_deps,
    session_deps=Deps,                     # enables load-before / save-after
    name="my-server",
)

@server.tool()
def remember(ctx: RunContext[Deps], key: str, value: str) -> str:
    ctx.deps.notes[key] = value            # mutate — auto-saved after this returns
    return f"Stored {key!r}"

server.run(transport="stdio")

from key_value.aio.stores.redis import RedisStore

server = MCPServer(
    deps=make_deps,
    session_deps=Deps,
    session_state_store=RedisStore(url=os.environ["REDIS_URL"]),
)
server.run(transport="streamable-http", host="0.0.0.0", port=8000)

from fastapi import FastAPI
from pydantic_ai_mcp import MCPServer

server = MCPServer(deps=make_deps, name="my-server")

@server.tool()
async def my_tool(ctx: RunContext[Deps]) -> str: ...

app = FastAPI()
app.mount("/mcp", server.http_app())

pip install git+https://github.com/thejens/pydantic-mcp-demo.git
# or
uv add git+https://github.com/thejens/pydantic-mcp-demo.git

@server.prompt
async def expert_prompt(ctx: RunContext[Deps], domain: str) -> str:
    """System prompt that makes the model a domain expert."""
    return f"You are an expert {domain} engineer."

def make_deps() -> Deps:
    return Deps(api_key=os.environ["API_KEY"])

async def make_deps_async() -> Deps:
    secret = await vault.get("api_key")
    return Deps(api_key=secret)

server = MCPServer(deps=make_deps, ...)

from key_value.aio.stores.redis import RedisStore

server = MCPServer(
    deps=make_deps,
    session_deps=Deps,
    session_state_store=RedisStore(url=os.environ["REDIS_URL"]),
)

server = MCPServer(
    toolsets=[search_toolset, files_toolset, metrics_toolset],
    deps=make_deps,
    prompts=[analyst_prompt, researcher_prompt],
    name="full-suite",
    instructions="Company intelligence tools.",
)
server.run(transport="stdio")

MCPServer(
    *,
    toolsets: Sequence[AbstractToolset[DepsT]] = (),
    deps: DepsT | Callable[[], DepsT] | Callable[[DepsT], DepsT] = None,
    session_deps: type[DepsT] | None = None,
    prompts: Sequence[Callable[..., Any]] | None = None,
    name: str = "pydantic-ai-mcp",
    bootstrap_deps: Any = None,
    **fastmcp_kwargs,
)