MCP API¶

mcp ¶

MCP integration for dynamic tool discovery and server exposure.

Provides MCPServerManager for connecting to MCP servers and using their tools as pydantic-ai tools, MCPToolAdapter for bridging MCP→pydantic-ai, and create_orqest_server for exposing Orqest as an MCP server.

Auto-discovery is gated by :class:PermissionGate (default :class:DenyAll); two integration paths are available:

:meth:ToolRegistry.get_or_discover — deliberate lookup (called by code that knows it needs a tool now).
:class:DiscoveryHook — opportunistic recovery from runtime "tool not found" errors raised by hallucinating LLMs.

MCPToolAdapter ¶

Convert MCP tool definitions into pydantic-ai Tool instances.

MCP tools are described by name, description, and a JSON Schema for their parameters. This adapter creates an async wrapper function that calls the MCP server and packages the result as a pydantic-ai Tool.

adapt `staticmethod` ¶

adapt(
    tool_name: str,
    tool_description: str,
    input_schema: dict[str, Any],
    call_fn: Any,
) -> Tool

Create a pydantic-ai Tool from a single MCP tool definition.

Parameters:

Name	Type	Description	Default
`tool_name`	`str`	The MCP tool name.	required
`tool_description`	`str`	Human-readable description shown to the LLM.	required
`input_schema`	`dict[str, Any]`	JSON Schema for the tool's input parameters.	required
`call_fn`	`Any`	Async callable `(tool_name, arguments_dict) -> result`.	required

Returns:

Type	Description
`Tool`	A pydantic-ai `Tool` wrapping the MCP call.

Source code in orqest/mcp/adapter.py

@staticmethod
def adapt(
    tool_name: str,
    tool_description: str,
    input_schema: dict[str, Any],
    call_fn: Any,
) -> Tool:
    """Create a pydantic-ai Tool from a single MCP tool definition.

    Args:
        tool_name: The MCP tool name.
        tool_description: Human-readable description shown to the LLM.
        input_schema: JSON Schema for the tool's input parameters.
        call_fn: Async callable ``(tool_name, arguments_dict) -> result``.

    Returns:
        A pydantic-ai ``Tool`` wrapping the MCP call.

    """

    async def _mcp_wrapper(**kwargs: Any) -> str:
        try:
            result = await call_fn(tool_name, kwargs)
            return _extract_text(result)
        except Exception as exc:
            logger.warning(
                "MCP tool {name} failed: {err}",
                name=tool_name,
                err=exc,
            )
            return f"Error calling {tool_name}: {exc}"

    return Tool(
        _mcp_wrapper,
        name=tool_name,
        description=tool_description,
    )

adapt_many `staticmethod` ¶

adapt_many(
    tool_defs: list[dict[str, Any]], call_fn: Any
) -> list[Tool]

Convert a list of MCP tool definitions to pydantic-ai Tools.

Each entry in tool_defs must have name and may have description and inputSchema keys (matching the MCP ListToolsResult shape).

Source code in orqest/mcp/adapter.py

@staticmethod
def adapt_many(
    tool_defs: list[dict[str, Any]],
    call_fn: Any,
) -> list[Tool]:
    """Convert a list of MCP tool definitions to pydantic-ai Tools.

    Each entry in *tool_defs* must have ``name`` and may have
    ``description`` and ``inputSchema`` keys (matching the MCP
    ``ListToolsResult`` shape).

    """
    adapted: list[Tool] = []
    for defn in tool_defs:
        name = defn.get("name", "")
        if not name:
            continue
        adapted.append(
            MCPToolAdapter.adapt(
                tool_name=name,
                tool_description=defn.get("description", ""),
                input_schema=defn.get("inputSchema", {}),
                call_fn=call_fn,
            )
        )
    return adapted

MCPConnection ¶

MCPConnection(config: MCPServerConfig)

A live connection to a single MCP server.

Manages the transport, session, and adapted tool list for one server.

Initialize with server config. Call connect() to start.

Source code in orqest/mcp/client.py

def __init__(self, config: MCPServerConfig) -> None:
    """Initialize with server config. Call ``connect()`` to start."""
    self.config = config
    self.name = config.name
    self._session: Any = None
    self._client_ctx: Any = None
    self._session_ctx: Any = None
    self._tools: list[Tool] = []
    self._connected = False

tools `property` ¶

tools: list[Tool]

All tools from this server.

connected `property` ¶

connected: bool

Whether the connection is alive.

tool_names `property` ¶

tool_names: list[str]

Names of all available tools.

connect `async` ¶

connect() -> None

Establish the MCP connection and discover tools.

Branches on config.transport: "stdio" launches the server process via stdio_client; "sse" connects to config.url via sse_client; "streamable-http" connects via streamablehttp_client with config.headers (the canonical transport for Tier-2 :class:DockerSandbox). The Streamable HTTP transport yields a (read, write, _session_id_callback) triple; the rest yield (read, write). We normalize on the first two.

Source code in orqest/mcp/client.py

async def connect(self) -> None:
    """Establish the MCP connection and discover tools.

    Branches on ``config.transport``: ``"stdio"`` launches the server
    process via ``stdio_client``; ``"sse"`` connects to ``config.url``
    via ``sse_client``; ``"streamable-http"`` connects via
    ``streamablehttp_client`` with ``config.headers`` (the canonical
    transport for Tier-2 :class:`DockerSandbox`). The Streamable HTTP
    transport yields a ``(read, write, _session_id_callback)`` triple;
    the rest yield ``(read, write)``. We normalize on the first two.
    """
    from mcp import ClientSession

    self._client_ctx = self._open_transport()
    transport_streams = await self._client_ctx.__aenter__()
    # Streamable HTTP returns (read, write, get_session_id); others (read, write)
    read, write = transport_streams[0], transport_streams[1]

    self._session_ctx = ClientSession(read, write)
    self._session = await self._session_ctx.__aenter__()
    await self._session.initialize()

    result = await self._session.list_tools()
    self._tools = MCPToolAdapter.adapt_many(
        [
            {
                "name": t.name,
                "description": t.description or "",
                "inputSchema": t.inputSchema,
            }
            for t in result.tools
        ],
        call_fn=self._call_tool,
    )
    self._connected = True
    logger.info(
        "Connected to MCP server '{name}' — {n} tools",
        name=self.name,
        n=len(self._tools),
    )

disconnect `async` ¶

disconnect() -> None

Close the connection gracefully.

Source code in orqest/mcp/client.py

async def disconnect(self) -> None:
    """Close the connection gracefully."""
    try:
        if self._session_ctx:
            await self._session_ctx.__aexit__(None, None, None)
        if self._client_ctx:
            await self._client_ctx.__aexit__(None, None, None)
    except Exception as exc:
        logger.debug(
            "Error disconnecting from {name}: {err}",
            name=self.name,
            err=exc,
        )
    finally:
        self._connected = False
        self._session = None
        self._tools = []

MCPServerManager ¶

MCPServerManager(config: MCPConfig | None = None)

Manage connections to multiple MCP servers.

Handles lifecycle, auto-discovery from standard config paths, and provides a unified tool list across all connected servers. Supports async with for automatic cleanup.

Initialize the manager.

Parameters:

Name	Type	Description	Default
`config`	`MCPConfig \| None`	Explicit MCP configuration. When auto_discover is `True` (the default), standard config paths are also scanned on `connect_all()`.	`None`

Source code in orqest/mcp/client.py

def __init__(self, config: MCPConfig | None = None) -> None:
    """Initialize the manager.

    Args:
        config: Explicit MCP configuration.  When *auto_discover* is
            ``True`` (the default), standard config paths are also
            scanned on ``connect_all()``.

    """
    self._config = config or MCPConfig()
    self._connections: dict[str, MCPConnection] = {}

connected_servers `property` ¶

connected_servers: list[str]

Names of all live connections.

total_tools `property` ¶

total_tools: int

Total tools across all connections.

connect_all `async` ¶

connect_all() -> None

Connect to all configured + discovered servers.

Source code in orqest/mcp/client.py

async def connect_all(self) -> None:
    """Connect to all configured + discovered servers."""
    configs = list(self._config.servers)
    if self._config.auto_discover:
        configs.extend(self.discover_local_configs())
    for cfg in configs:
        try:
            await self.connect(cfg)
        except Exception:
            pass  # logged inside connect()

connect `async` ¶

connect(config: MCPServerConfig) -> MCPConnection

Connect to a single MCP server.

Raises:

Type	Description
`Exception`	Propagated from the transport layer on failure.

Source code in orqest/mcp/client.py

async def connect(self, config: MCPServerConfig) -> MCPConnection:
    """Connect to a single MCP server.

    Raises:
        Exception: Propagated from the transport layer on failure.

    """
    conn = MCPConnection(config)
    try:
        await conn.connect()
        self._connections[config.name] = conn
        return conn
    except Exception as exc:
        logger.warning(
            "Failed to connect to MCP server '{name}': {err}",
            name=config.name,
            err=exc,
        )
        raise

disconnect_all `async` ¶

disconnect_all() -> None

Disconnect from every server.

Source code in orqest/mcp/client.py

async def disconnect_all(self) -> None:
    """Disconnect from every server."""
    for conn in self._connections.values():
        await conn.disconnect()
    self._connections.clear()

disconnect `async` ¶

disconnect(name: str) -> None

Disconnect from a specific server by name.

Source code in orqest/mcp/client.py

async def disconnect(self, name: str) -> None:
    """Disconnect from a specific server by name."""
    conn = self._connections.pop(name, None)
    if conn:
        await conn.disconnect()

discover_and_connect `async` ¶

discover_and_connect(
    query: str, *, max_servers: int = 3
) -> list[MCPConnection]

Search online for MCP servers matching a capability and connect.

Uses MCPDiscovery to find servers by keyword, then connects to each. This is the key method that enables agents to dynamically expand their toolset at runtime.

Parameters:

Name	Type	Description	Default
`query`	`str`	Capability description (e.g., "SQL database", "GitHub").	required
`max_servers`	`int`	Maximum number of servers to connect to.	`3`

Returns:

Type	Description
`list[MCPConnection]`	List of newly established connections.

Source code in orqest/mcp/client.py

async def discover_and_connect(
    self,
    query: str,
    *,
    max_servers: int = 3,
) -> list[MCPConnection]:
    """Search online for MCP servers matching a capability and connect.

    Uses ``MCPDiscovery`` to find servers by keyword, then connects
    to each. This is the key method that enables agents to
    dynamically expand their toolset at runtime.

    Args:
        query: Capability description (e.g., "SQL database", "GitHub").
        max_servers: Maximum number of servers to connect to.

    Returns:
        List of newly established connections.

    """
    from orqest.mcp.discovery import MCPDiscovery

    discovery = MCPDiscovery()
    discovered = await discovery.search(query, max_results=max_servers)

    new_connections: list[MCPConnection] = []
    for server in discovered:
        if server.name in self._connections:
            continue  # Already connected
        try:
            config = server.to_config()
            conn = await self.connect(config)
            new_connections.append(conn)
            logger.info(
                "Discovered and connected to '{name}' for '{q}'",
                name=server.name,
                q=query,
            )
        except Exception:
            pass  # Logged inside connect()

    return new_connections

get_all_tools ¶

get_all_tools() -> list[Tool]

Collect tools from all live connections.

Source code in orqest/mcp/client.py

def get_all_tools(self) -> list[Tool]:
    """Collect tools from all live connections."""
    tools: list[Tool] = []
    for conn in self._connections.values():
        if conn.connected:
            tools.extend(conn.tools)
    return tools

get_tools ¶

get_tools(server_name: str) -> list[Tool]

Get tools from one server.

Source code in orqest/mcp/client.py

def get_tools(self, server_name: str) -> list[Tool]:
    """Get tools from one server."""
    conn = self._connections.get(server_name)
    if conn and conn.connected:
        return conn.tools
    return []

search_tools ¶

search_tools(query: str) -> list[Tool]

Keyword search across all connected servers.

Source code in orqest/mcp/client.py

def search_tools(self, query: str) -> list[Tool]:
    """Keyword search across all connected servers."""
    q = query.lower()
    return [
        t
        for t in self.get_all_tools()
        if q in t.name.lower()
        or q in (getattr(t, "description", "") or "").lower()
    ]

discover_local_configs `staticmethod` ¶

discover_local_configs() -> list[MCPServerConfig]

Scan standard paths for MCP server configurations.

Checks ~/.claude.json, ~/.claude/claude.json, and ~/.config/Claude/claude_desktop_config.json.

Source code in orqest/mcp/client.py

@staticmethod
def discover_local_configs() -> list[MCPServerConfig]:
    """Scan standard paths for MCP server configurations.

    Checks ``~/.claude.json``, ``~/.claude/claude.json``, and
    ``~/.config/Claude/claude_desktop_config.json``.
    """
    configs: list[MCPServerConfig] = []
    seen_names: set[str] = set()
    search_paths = [
        Path.home() / ".claude.json",
        Path.home() / ".claude" / "claude.json",
        Path.home() / ".config" / "Claude" / "claude_desktop_config.json",
    ]
    for path in search_paths:
        if not path.exists():
            continue
        try:
            data = json.loads(path.read_text())
            for name, defn in data.get("mcpServers", {}).items():
                if name in seen_names:
                    continue
                seen_names.add(name)
                configs.append(
                    MCPServerConfig(
                        name=name,
                        command=defn.get("command", ""),
                        args=defn.get("args", []),
                        env=defn.get("env", {}),
                    )
                )
            if configs:
                logger.debug(
                    "Discovered {n} MCP servers from {p}",
                    n=len(configs),
                    p=path,
                )
        except Exception as exc:
            logger.debug(
                "Could not read {p}: {e}", p=path, e=exc
            )
    return configs

MCPConfig `dataclass` ¶

MCPConfig(
    servers: list[MCPServerConfig] = list(),
    auto_discover: bool = True,
)

Top-level MCP integration configuration.

Parameters:

Name	Type	Description	Default
`servers`	`list[MCPServerConfig]`	Explicit server configurations.	`list()`
`auto_discover`	`bool`	Scan standard paths for MCP configs on connect.	`True`

MCPServerConfig `dataclass` ¶

MCPServerConfig(
    name: str,
    command: str,
    args: list[str] = list(),
    env: dict[str, str] = dict(),
    transport: str = "stdio",
    url: str | None = None,
    headers: dict[str, str] = dict(),
)

Configuration for a single MCP server connection.

Parameters:

Name	Type	Description	Default
`name`	`str`	Human-readable identifier for this server.	required
`command`	`str`	Executable to launch (e.g., "python", "node", "npx").	required
`args`	`list[str]`	Command-line arguments (e.g., ["-m", "my_mcp_server"]).	`list()`
`env`	`dict[str, str]`	Environment variables passed to the server process.	`dict()`
`transport`	`str`	Connection transport — `"stdio"`, `"sse"`, or `"streamable-http"`. `"streamable-http"` is the canonical transport for host↔container deployments (used by Tier-2 :class:`DockerSandbox`); `"sse"` is the deprecated 2024-11-05 transport (still supported for legacy MCP servers); `"stdio"` launches the server as a subprocess.	`'stdio'`
`url`	`str \| None`	Server URL for `"sse"` and `"streamable-http"` transports. Ignored for stdio. For `"streamable-http"` should be the `/mcp` endpoint (e.g. `http://127.0.0.1:8000/mcp`).	`None`
`headers`	`dict[str, str]`	HTTP headers attached to every request. Used to carry session/auth tokens (e.g. `{"Authorization": "Bearer <jwt>"}`). Empty for stdio.	`dict()`

DiscoveredServer `dataclass` ¶

DiscoveredServer(
    name: str,
    description: str,
    url: str,
    tools: list[str] = list(),
    source: str = "registry",
    metadata: dict[str, Any] = dict(),
)

A server found via online discovery.

Attributes:

Name	Type	Description
`name`	`str`	Server identifier.
`description`	`str`	What the server provides.
`url`	`str`	Connection endpoint (SSE or Streamable HTTP).
`tools`	`list[str]`	Tool names advertised by the server.
`source`	`str`	Where this was discovered ("registry", "wellknown", "search").
`metadata`	`dict[str, Any]`	Additional discovery metadata.

to_config ¶

to_config() -> MCPServerConfig

Convert to an MCPServerConfig for connection.

Source code in orqest/mcp/discovery.py

def to_config(self) -> MCPServerConfig:
    """Convert to an MCPServerConfig for connection."""
    return MCPServerConfig(
        name=self.name,
        command="",  # Not needed for SSE/HTTP transport
        transport="sse",
        url=self.url,
    )

MCPDiscovery ¶

MCPDiscovery(
    *,
    registry_urls: list[str] | None = None,
    well_known_urls: list[str] | None = None,
    timeout: float = 15.0,
)

Discover MCP servers online by querying registry endpoints.

Enables agents to find tools they need at runtime without any pre-configuration. The MetaOrchestrator uses this when no local tool matches a subtask's requirements. Preview — see the module docstring for current limitations.

Usage::

discovery = MCPDiscovery()
servers = await discovery.search("database SQL query")
for server in servers:
    config = server.to_config()
    await manager.connect(config)

Initialize discovery.

Parameters:

Name	Type	Description	Default
`registry_urls`	`list[str] \| None`	Custom registry search endpoints. Defaults to the official MCP registry and Glama.	`None`
`well_known_urls`	`list[str] \| None`	Base URLs to probe for `/.well-known/mcp.json` manifests on every :meth:`search`. Empty by default.	`None`
`timeout`	`float`	HTTP request timeout in seconds.	`15.0`

Source code in orqest/mcp/discovery.py

def __init__(
    self,
    *,
    registry_urls: list[str] | None = None,
    well_known_urls: list[str] | None = None,
    timeout: float = 15.0,
) -> None:
    """Initialize discovery.

    Args:
        registry_urls: Custom registry search endpoints.
            Defaults to the official MCP registry and Glama.
        well_known_urls: Base URLs to probe for ``/.well-known/mcp.json``
            manifests on every :meth:`search`. Empty by default.
        timeout: HTTP request timeout in seconds.

    """
    self._registry_urls = registry_urls or list(self.REGISTRY_SEARCH_URLS)
    self._well_known_urls = list(well_known_urls or [])
    self._timeout = timeout

search `async` ¶

search(
    query: str, *, max_results: int = 5
) -> list[DiscoveredServer]

Search for MCP servers matching a capability query.

Probes any configured well_known_urls first (explicitly configured = highest intent), then queries the registry endpoints, deduplicating by server name.

Parameters:

Name	Type	Description	Default
`query`	`str`	Natural language description of needed capability (e.g., "SQL database queries", "GitHub repository access").	required
`max_results`	`int`	Maximum servers to return.	`5`

Returns:

Type	Description
`list[DiscoveredServer]`	Discovered servers, well-known manifests ahead of registry hits.

Source code in orqest/mcp/discovery.py

async def search(
    self,
    query: str,
    *,
    max_results: int = 5,
) -> list[DiscoveredServer]:
    """Search for MCP servers matching a capability query.

    Probes any configured ``well_known_urls`` first (explicitly
    configured = highest intent), then queries the registry endpoints,
    deduplicating by server name.

    Args:
        query: Natural language description of needed capability
            (e.g., "SQL database queries", "GitHub repository access").
        max_results: Maximum servers to return.

    Returns:
        Discovered servers, well-known manifests ahead of registry hits.

    """
    all_servers: list[DiscoveredServer] = []
    seen_names: set[str] = set()

    # Explicitly-configured well-known manifests first — highest intent.
    for base_url in self._well_known_urls:
        try:
            server = await self.probe_wellknown(base_url)
        except Exception as exc:
            logger.debug(
                "Well-known probe {url} failed: {err}", url=base_url, err=exc
            )
            continue
        if server is not None and server.name not in seen_names:
            seen_names.add(server.name)
            all_servers.append(server)

    # Then fuzzy registry search.
    for url in self._registry_urls:
        try:
            results = await self._query_registry(url, query, max_results)
            for server in results:
                if server.name not in seen_names:
                    seen_names.add(server.name)
                    all_servers.append(server)
        except Exception as exc:
            logger.debug(
                "Registry {url} query failed: {err}",
                url=url,
                err=exc,
            )

    return all_servers[:max_results]

probe_wellknown `async` ¶

probe_wellknown(base_url: str) -> DiscoveredServer | None

Probe a URL's /.well-known/mcp.json for server metadata.

Parameters:

Name	Type	Description	Default
`base_url`	`str`	The server's base URL (e.g., "https://mcp.example.com").	required

Returns:

Type	Description
`DiscoveredServer \| None`	A DiscoveredServer if the manifest exists, else None.

Source code in orqest/mcp/discovery.py

async def probe_wellknown(self, base_url: str) -> DiscoveredServer | None:
    """Probe a URL's ``/.well-known/mcp.json`` for server metadata.

    Args:
        base_url: The server's base URL (e.g., "https://mcp.example.com").

    Returns:
        A DiscoveredServer if the manifest exists, else None.

    """
    try:
        import httpx

        wellknown_url = f"{base_url.rstrip('/')}/.well-known/mcp.json"
        async with httpx.AsyncClient(timeout=self._timeout) as client:
            resp = await client.get(wellknown_url)
            if resp.status_code != 200:
                return None
            data = resp.json()
            return DiscoveredServer(
                name=data.get("name", base_url),
                description=data.get("description", ""),
                url=data.get("endpoint", base_url),
                tools=[
                    t.get("name", "")
                    for t in data.get("tools", [])
                    if t.get("name")
                ],
                source="wellknown",
                metadata=data,
            )
    except Exception as exc:
        logger.debug(
            "Well-known probe failed for {url}: {err}",
            url=base_url,
            err=exc,
        )
        return None

DiscoveryHook ¶

DiscoveryHook(
    registry: ToolRegistry,
    discovery: MCPDiscovery,
    manager: MCPServerManager,
    *,
    permission: PermissionGate | None = None,
    audit_bus: EventBus | None = None,
)

ToolHook that recovers from "tool not found" via MCP discovery.

The hook's :meth:on_error returns :class:Redirect(new_tool=name) after the tool is registered (caller should retry with the registered tool), or :class:Continue otherwise — including when the gate denies or discovery fails.

Source code in orqest/mcp/discovery_hook.py

def __init__(
    self,
    registry: ToolRegistry,
    discovery: MCPDiscovery,
    manager: MCPServerManager,
    *,
    permission: PermissionGate | None = None,
    audit_bus: EventBus | None = None,
) -> None:
    self._registry = registry
    self._discovery = discovery
    self._manager = manager
    self._permission = permission
    self._audit_bus = audit_bus

AllowAll ¶

Permit any tool name. Use only for development / trusted environments.

AllowList ¶

AllowList(patterns: list[str])

Permit tool names matching any of the supplied regex patterns.

Patterns are compiled with :func:re.search semantics — they match anywhere in the tool name. Anchor with ^…$ for full-name matches.

Source code in orqest/mcp/permission.py

def __init__(self, patterns: list[str]) -> None:
    self._patterns = [re.compile(p) for p in patterns]

DenyAll ¶

Deny every discovery. The default — discovery is opt-in.

PermissionGate ¶

Decide whether a tool name may be discovered + registered at runtime.

MCP API¶

mcp ¶

MCPToolAdapter ¶

adapt staticmethod ¶

adapt_many staticmethod ¶

MCPConnection ¶

tools property ¶

connected property ¶

tool_names property ¶

connect async ¶

disconnect async ¶

MCPServerManager ¶

connected_servers property ¶

total_tools property ¶

connect_all async ¶

connect async ¶

disconnect_all async ¶

disconnect async ¶

discover_and_connect async ¶

get_all_tools ¶

get_tools ¶

search_tools ¶

discover_local_configs staticmethod ¶

MCPConfig dataclass ¶

MCPServerConfig dataclass ¶

DiscoveredServer dataclass ¶

to_config ¶

MCPDiscovery ¶

search async ¶

probe_wellknown async ¶

DiscoveryHook ¶

AllowAll ¶

AllowList ¶

DenyAll ¶

PermissionGate ¶

adapt `staticmethod` ¶

adapt_many `staticmethod` ¶

tools `property` ¶

connected `property` ¶

tool_names `property` ¶

connect `async` ¶

disconnect `async` ¶

connected_servers `property` ¶

total_tools `property` ¶

connect_all `async` ¶

connect `async` ¶

disconnect_all `async` ¶

disconnect `async` ¶

discover_and_connect `async` ¶

discover_local_configs `staticmethod` ¶

MCPConfig `dataclass` ¶

MCPServerConfig `dataclass` ¶

DiscoveredServer `dataclass` ¶

search `async` ¶

probe_wellknown `async` ¶