Skip to main content

mcp.toolsets

MCP Toolsets management for Pydantic AI agents.

This module provides server-level MCP toolset management that loads MCP servers once at server startup and makes them available to all agents.

MCP servers are managed using the MCPLifecycleManager which handles:

  • Starting/stopping MCP servers dynamically
  • Tracking running state and tool discovery
  • Merging mcp.json config with catalog commands

Uses Pydantic AI's built-in MCP client support (MCPServerStdio, MCPServerStreamableHTTP) which automatically detects the transport type from the config:

  • command field → MCPServerStdio (stdio transport)
  • url field ending with /sse → MCPServerSSE (deprecated SSE transport)
  • url field (not /sse) → MCPServerStreamableHTTP (recommended HTTP transport)

MCP_SERVER_STARTUP_TIMEOUT

5 minutes

get_mcp_config_path

def get_mcp_config_path() -> Path

Get the path to the MCP configuration file.

Returns:

Path to mcp.json file

ensure_config_mcp_toolsets_event

def ensure_config_mcp_toolsets_event() -> None

Ensure the initialization event exists for external waiters.

initialize_config_mcp_toolsets

async def initialize_config_mcp_toolsets() -> None

Initialize MCP toolsets at server startup.

This loads MCP servers from the config file and starts them. For each server in mcp.json:

  • If it matches a catalog server, use the catalog command
  • Otherwise use the command from mcp.json

Respects the AGENT_RUNTIMES_NO_CONFIG_MCP_SERVERS environment variable: if set to "true", skips initialization entirely.

shutdown_config_mcp_toolsets

async def shutdown_config_mcp_toolsets() -> None

Shutdown config MCP toolsets at server shutdown.

This stops all running MCP server connections/subprocesses.

get_config_mcp_toolsets

def get_config_mcp_toolsets() -> list[Any]

Get the list of successfully started config MCP toolsets.

These can be passed directly to Pydantic AI Agent(toolsets=...).

Returns:

List of MCP server toolsets

get_config_mcp_toolsets_status

def get_config_mcp_toolsets_status() -> dict[str, Any]

Get the status of config MCP toolsets initialization.

Returns:

Dict with status information including:

  • initialized: Whether initialization has completed
  • ready_count: Number of servers ready
  • failed_count: Number of servers that failed to start
  • ready_servers: List of server IDs that are ready
  • failed_servers: Dict of server ID -> error message for failed servers
  • servers: List of per-server status dicts with id, status, error, tools_count

wait_for_config_mcp_toolsets

async def wait_for_config_mcp_toolsets(timeout: float | None = None) -> bool

Wait until config MCP toolsets initialization completes.

get_config_mcp_toolsets_info

def get_config_mcp_toolsets_info() -> list[dict[str, Any]]

Get information about the loaded config MCP toolsets.

Returns:

List of dicts with toolset info (type, id, command/url)

  • Note - Sensitive information like cookies/tokens in args are redacted.

is_config_mcp_toolsets_initialized

def is_config_mcp_toolsets_initialized() -> bool

Return True when config MCP toolsets initialization has completed.