Skip to main content

mcp.lifecycle

MCP Server Lifecycle Manager.

Centralized manager for MCP server lifecycle (start/stop/state tracking). Integrates with the MCP catalog to use predefined commands when available.

MCP_SERVER_STARTUP_TIMEOUT​

5 minutes

MCPServerInstance Objects​

class MCPServerInstance()

Represents a running MCP server instance.

MCPLifecycleManager Objects​

class MCPLifecycleManager()

Centralized manager for MCP server lifecycle.

Handles:

  • Starting/stopping MCP servers
  • Tracking running state (separately for config and catalog servers)
  • Merging mcp.json config with catalog commands
  • Tool discovery from running servers

Config servers (from mcp.json) and catalog servers (predefined) are managed in separate data structures so the same server ID can exist in both.

__init__​

def __init__() -> None

Initialize the lifecycle manager.

get_mcp_config_path​

def get_mcp_config_path() -> Path

Get the path to the MCP configuration file.

get_server_config_from_file​

def get_server_config_from_file(server_id: str) -> MCPServer | None

Get config specifically from mcp.json for a server.

Arguments:

  • server_id - The server identifier to look up in mcp.json

Returns:

MCPServer config if found in mcp.json, None otherwise. The returned config will have is_config=True.

get_merged_server_config​

def get_merged_server_config(
        server_id: str,
        user_config: dict[str, Any] | None = None,
        from_config_file: bool = False) -> MCPServer | None

Get server config, merging mcp.json with catalog if available.

Priority:

  1. If user_config has a 'command', use user config entirely (they know what they want)
  2. If server_id matches a catalog server and no user command, use catalog command
  3. Env vars from user_config are always merged in

Arguments:

  • server_id - The server identifier
  • user_config - Optional user-provided config overrides
  • from_config_file - True if this server is from mcp.json (marks as config server)

Returns:

MCPServer config or None if not found

start_server​

async def start_server(
        server_id: str,
        config: MCPServer | None = None,
        extra_env: dict[str, str] | None = None) -> MCPServerInstance | None

Start an MCP server.

Arguments:

  • server_id - The server identifier
  • config - Optional MCPServer config. If not provided, will try to get from catalog or mcp.json.
  • extra_env - Additional environment variables to pass to the server subprocess. These are merged on top of os.environ and config.env, and are also available for ${VAR} expansion in server args and config env values.

Returns:

MCPServerInstance if started successfully, None otherwise

stop_server​

async def stop_server(server_id: str, is_config: bool = False) -> bool

Stop a running MCP server.

Arguments:

  • server_id - The server identifier
  • is_config - Whether this is a config server (from mcp.json) or catalog server

Returns:

True if stopped successfully, False otherwise

get_running_server​

def get_running_server(
        server_id: str,
        is_config: bool | None = None) -> MCPServerInstance | None

Get a running server instance by ID.

Arguments:

  • server_id - The server identifier
  • is_config - If True, only check config servers. If False, only check catalog. If None, check both (catalog first, then config).

get_config_servers​

def get_config_servers() -> list[MCPServerInstance]

Get all running config server instances (from mcp.json).

get_catalog_servers​

def get_catalog_servers() -> list[MCPServerInstance]

Get all running catalog server instances.

get_all_running_servers​

def get_all_running_servers() -> list[MCPServerInstance]

Get all running server instances (both config and catalog).

get_running_server_ids​

def get_running_server_ids() -> list[str]

Get IDs of all running servers (combined, may have duplicates).

get_config_server_ids​

def get_config_server_ids() -> list[str]

Get IDs of running config servers.

get_catalog_server_ids​

def get_catalog_server_ids() -> list[str]

Get IDs of running catalog servers.

is_server_running​

def is_server_running(server_id: str, is_config: bool | None = None) -> bool

Check if a server is running.

Arguments:

  • server_id - The server identifier
  • is_config - If True, only check config servers. If False, only check catalog. If None, check both.

is_config_server_running​

def is_config_server_running(server_id: str) -> bool

Check if a config server (from mcp.json) is running.

is_catalog_server_running​

def is_catalog_server_running(server_id: str) -> bool

Check if a catalog server is running.

get_failed_servers​

def get_failed_servers() -> dict[str, str]

Get dict of failed server IDs to error messages.

get_server_status​

def get_server_status(server_id: str,
                      is_config: bool | None = None) -> dict[str, Any]

Get status of a specific server.

get_all_servers_status​

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

Get per-server status for all known servers.

Each entry has:

  • id: server identifier
  • status: one of "none", "not_started", "starting", "failed", "started"
  • error: error message (only when status == "failed")
  • tools_count: number of discovered tools (only when status == "started")

get_pydantic_toolsets​

def get_pydantic_toolsets(include_config: bool = True,
                          include_catalog: bool = True) -> list[Any]

Get running MCP servers as pydantic_ai toolsets.

Arguments:

  • include_config - Whether to include config servers (from mcp.json)
  • include_catalog - Whether to include catalog servers

Returns:

List of pydantic MCP server instances for use with Agent(toolsets=...)

initialize_from_config​

async def initialize_from_config() -> None

Initialize MCP servers from mcp.json config file.

For each server in the config:

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

shutdown​

async def shutdown() -> None

Shutdown all running MCP servers.

wait_for_initialization​

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

Wait for initialization to complete.

is_initialized​

def is_initialized() -> bool

Check if initialization has completed.

get_mcp_lifecycle_manager​

def get_mcp_lifecycle_manager() -> MCPLifecycleManager

Get the global MCP lifecycle manager instance.

set_mcp_lifecycle_manager​

def set_mcp_lifecycle_manager(manager: MCPLifecycleManager) -> None

Set the global MCP lifecycle manager instance.