Skip to main content

commands.agent_mcp_servers

Agent MCP servers commands for the Agent Runtimes CLI.

This module provides commands to start and stop MCP servers for running agents.

Usage as library: ```python from agent_runtimes.commands.agent_mcp_servers import ( start_agent_mcp_servers, stop_agent_mcp_servers, parse_env_vars, )

# Start MCP servers for a specific agent
result = start_agent_mcp_servers(
    agent_id="my-agent",
    env_vars={"TAVILY_API_KEY": "xxx"},
    host="127.0.0.1",
    port=8000,
)

# Start MCP servers for all agents (agent_id=None)
result = start_agent_mcp_servers(
    agent_id=None,
    env_vars={"TAVILY_API_KEY": "xxx"},
    host="127.0.0.1",
    port=8000,
)

# Stop MCP servers
result = stop_agent_mcp_servers(
    agent_id="my-agent",  # or None for all agents
    host="127.0.0.1",
    port=8000,
)
```

AgentMcpServersError Objects​

class AgentMcpServersError(Exception)

Error during agent MCP server operations.

parse_env_vars​

def parse_env_vars(env_vars_str: str | None) -> dict[str, str]

Parse environment variables from a semicolon-separated string.

Arguments:

  • env_vars_str - String in format "VAR1:VALUE1;VAR2:VALUE2"

Returns:

Dictionary of environment variable name-value pairs.

Examples:

>>> parse_env_vars("TAVILY_API_KEY:xxx;OTHER_KEY:yyy")

  • {'TAVILY_API_KEY' - 'xxx', 'OTHER_KEY': 'yyy'} >>> parse_env_vars(None) {}

start_agent_mcp_servers​

def start_agent_mcp_servers(agent_id: str | None = None,
                            env_vars: dict[str, str] | None = None,
                            host: str = "127.0.0.1",
                            port: int = 8000,
                            timeout: float = 60.0) -> dict[str, Any]

Start MCP servers for a running agent or all agents.

Arguments:

  • agent_id - The agent identifier. If None, operates on all agents.
  • env_vars - Environment variables to set before starting servers.
  • host - Server host.
  • port - Server port.
  • timeout - Request timeout in seconds.

Returns:

Response from the server with status of each server.

Raises:

  • AgentMcpServersError - If the request fails.

stop_agent_mcp_servers​

def stop_agent_mcp_servers(agent_id: str | None = None,
                           host: str = "127.0.0.1",
                           port: int = 8000,
                           timeout: float = 60.0) -> dict[str, Any]

Stop MCP servers for a running agent or all agents.

Arguments:

  • agent_id - The agent identifier. If None, operates on all agents.
  • host - Server host.
  • port - Server port.
  • timeout - Request timeout in seconds.

Returns:

Response from the server with status of each server.

Raises:

  • AgentMcpServersError - If the request fails.
def print_mcp_servers_result(result: dict[str, Any],
                             operation: str = "start") -> None

Print MCP server operation result in a formatted way.

Arguments:

  • result - Response from start/stop operation.
  • operation - Either "start" or "stop".