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:
commandfield → MCPServerStdio (stdio transport)urlfield ending with/sse→ MCPServerSSE (deprecated SSE transport)urlfield (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.