Skip to content

karenina.utils.mcp

mcp

Shared MCP utilities for Karenina.

This module provides adapter-agnostic MCP utilities using the core mcp package. For adapter-specific MCP functionality: - LangChain tools: karenina.adapters.langchain.mcp - Claude SDK tools: karenina.adapters.claude_tool.mcp

The multi-server helpers (connect_all_mcp_servers, get_all_mcp_tools) that used to live here were unused duplicates of the live copies in karenina.adapters.claude_tool.mcp and were removed. Use those instead.

Classes

ManagedMcpServer dataclass

A running managed MCP server.

Source code in src/karenina/utils/mcp/server.py 
72
73
74
75
76
77
78
79
80
81
82
@dataclass(frozen=True)
class ManagedMcpServer:
    """A running managed MCP server."""

    spec: McpServerSpec
    process: subprocess.Popen[str]

    @property
    def url(self) -> str:
        """Return the server's streamable HTTP URL."""
        return self.spec.url
Attributes
url property 
url: str

Return the server's streamable HTTP URL.

McpServerSpec dataclass

Configuration for one managed local MCP server.

Parameters:

Name Type Description Default
name
str

Human-readable server name used in logs and errors.

 required
command
tuple[str, ...]

Executable and arguments used to start the server.

 required
host
str

Host used for the readiness check and returned URL.

 '127.0.0.1'
port
int

TCP port used for the readiness check and returned URL.

 8000
path
str

HTTP path appended to the returned URL.

 '/mcp'
env
Mapping[str, str]

Environment additions passed to the child process.

 dict()
cwd
Path | None

Optional child working directory.

 None
log_path
Path | None

Optional file receiving combined standard output and error.

 None
startup_timeout
float

Seconds to wait for the TCP port to accept connections.

 60.0
shutdown_timeout
float

Seconds to wait after termination before forcing kill.

 10.0
poll_interval
float

Seconds between readiness checks.

 0.1
Source code in src/karenina/utils/mcp/server.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@dataclass(frozen=True)
class McpServerSpec:
    """Configuration for one managed local MCP server.

    Args:
        name: Human-readable server name used in logs and errors.
        command: Executable and arguments used to start the server.
        host: Host used for the readiness check and returned URL.
        port: TCP port used for the readiness check and returned URL.
        path: HTTP path appended to the returned URL.
        env: Environment additions passed to the child process.
        cwd: Optional child working directory.
        log_path: Optional file receiving combined standard output and error.
        startup_timeout: Seconds to wait for the TCP port to accept connections.
        shutdown_timeout: Seconds to wait after termination before forcing kill.
        poll_interval: Seconds between readiness checks.
    """

    name: str
    command: tuple[str, ...]
    host: str = "127.0.0.1"
    port: int = 8000
    path: str = "/mcp"
    env: Mapping[str, str] = field(default_factory=dict)
    cwd: Path | None = None
    log_path: Path | None = None
    startup_timeout: float = 60.0
    shutdown_timeout: float = 10.0
    poll_interval: float = 0.1

    def __post_init__(self) -> None:
        """Validate values needed for safe process management."""
        if not self.name.strip():
            raise ValueError("MCP server name must not be empty")
        if not self.command or not self.command[0].strip():
            raise ValueError("MCP server command must not be empty")
        if not 1 <= self.port <= 65_535:
            raise ValueError("MCP server port must be between 1 and 65535")
        if self.startup_timeout <= 0 or self.shutdown_timeout <= 0:
            raise ValueError("MCP server timeouts must be positive")
        if self.poll_interval <= 0:
            raise ValueError("MCP server poll interval must be positive")

    @property
    def url(self) -> str:
        """Return the server's streamable HTTP URL."""
        path = self.path if self.path.startswith("/") else f"/{self.path}"
        return f"http://{self.host}:{self.port}{path}"
Attributes
url property 
url: str

Return the server's streamable HTTP URL.

Functions

afetch_tool_descriptions async 

afetch_tool_descriptions(
    mcp_urls_dict: dict[str, str],
    tool_filter: list[str] | None = None,
) -> dict[str, str]

Fetch tool descriptions from MCP servers using the core mcp package.

This function connects to MCP servers and retrieves the descriptions for all available tools. Useful for getting seed descriptions for GEPA optimization.

Parameters:

Name Type Description Default
mcp_urls_dict
dict[str, str]

Dictionary mapping server names to MCP server URLs

 required
tool_filter
list[str] | None

Optional list of tool names to include

 None

Returns:

Type Description
dict[str, str]

Dict mapping tool names to their descriptions
Example

descriptions = await fetch_tool_descriptions( ... {"biocontext": "https://mcp.biocontext.ai/mcp/"} ... ) print(descriptions) {'search_proteins': 'Search for protein information...', ...}

Source code in src/karenina/utils/mcp/tools.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
async def afetch_tool_descriptions(
    mcp_urls_dict: dict[str, str],
    tool_filter: list[str] | None = None,
) -> dict[str, str]:
    """Fetch tool descriptions from MCP servers using the core mcp package.

    This function connects to MCP servers and retrieves the descriptions
    for all available tools. Useful for getting seed descriptions for
    GEPA optimization.

    Args:
        mcp_urls_dict: Dictionary mapping server names to MCP server URLs
        tool_filter: Optional list of tool names to include

    Returns:
        Dict mapping tool names to their descriptions

    Example:
        >>> descriptions = await fetch_tool_descriptions(
        ...     {"biocontext": "https://mcp.biocontext.ai/mcp/"}
        ... )
        >>> print(descriptions)
        {'search_proteins': 'Search for protein information...', ...}
    """
    from mcp import ClientSession
    from mcp.client.streamable_http import streamablehttp_client

    from karenina.adapters._timeouts import MCP_CONNECT_TIMEOUT

    descriptions: dict[str, str] = {}

    async with AsyncExitStack() as stack:
        for server_name, url in mcp_urls_dict.items():
            try:
                # Connect to MCP server. The inner per-server bound is
                # derived from the same source as the outer dispatch bound
                # in fetch_tool_descriptions and fires first for a single
                # server. With multiple servers the per-server bounds sum,
                # so the outer dispatch bound may fire first instead.
                http_transport = await asyncio.wait_for(
                    stack.enter_async_context(streamablehttp_client(url)),
                    timeout=MCP_CONNECT_TIMEOUT,
                )
                read_stream, write_stream, _ = http_transport

                session = await stack.enter_async_context(ClientSession(read_stream, write_stream))
                await session.initialize()

                # List tools from this server
                tools_response = await session.list_tools()
                mcp_tools = tools_response.tools

                # Extract descriptions
                for tool in mcp_tools:
                    tool_name = tool.name
                    tool_desc = tool.description or ""

                    # Apply filter if provided
                    if tool_filter is not None and tool_name not in tool_filter:
                        continue

                    descriptions[tool_name] = tool_desc

                logger.debug(f"Fetched {len(mcp_tools)} tools from MCP server '{server_name}'")

            except TimeoutError as e:
                raise McpTimeoutError(
                    f"MCP server '{server_name}' connection timed out after {MCP_CONNECT_TIMEOUT:.0f} seconds.",
                    server_name=server_name,
                    timeout_seconds=int(MCP_CONNECT_TIMEOUT),
                ) from e
            except Exception as e:
                logger.error(f"Failed to fetch tools from MCP server '{server_name}': {e}")
                raise

    logger.debug(f"Fetched descriptions for {len(descriptions)} total tools")
    return descriptions

apply_tool_description_overrides 

apply_tool_description_overrides(
    tools: list[Any], overrides: dict[str, str]
) -> list[Any]

Apply description overrides to tools.

Used by GEPA optimization to test different tool descriptions.

Parameters:

Name Type Description Default
tools
list[Any]

List of tool objects (LangChain Tool or MCP Tool)

 required
overrides
dict[str, str]

Dict mapping tool names to new descriptions

 required

Returns:

Type Description
list[Any]

Modified tools list with updated descriptions
Source code in src/karenina/utils/mcp/tools.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def apply_tool_description_overrides(
    tools: list[Any],
    overrides: dict[str, str],
) -> list[Any]:
    """Apply description overrides to tools.

    Used by GEPA optimization to test different tool descriptions.

    Args:
        tools: List of tool objects (LangChain Tool or MCP Tool)
        overrides: Dict mapping tool names to new descriptions

    Returns:
        Modified tools list with updated descriptions
    """
    for tool in tools:
        tool_name = getattr(tool, "name", None)
        if tool_name and tool_name in overrides:
            tool.description = overrides[tool_name]
    return tools

connect_mcp_session async 

connect_mcp_session(
    exit_stack: AsyncExitStack, config: MCPServerConfig
) -> ClientSession

Connect to an MCP server via HTTP/SSE transport.

Creates a new MCP client session and registers it with the exit stack for automatic cleanup when the stack closes.

Parameters:

Name Type Description Default
exit_stack
AsyncExitStack

AsyncExitStack for managing session lifecycle.

 required
config
MCPServerConfig

MCP server configuration. Must include 'url' for HTTP/SSE transport. Optional 'headers' dict for authentication. Optional 'timeout' and 'sse_read_timeout' floats (seconds) override the MCP SDK defaults.

 required

Returns:

Type Description
ClientSession

Initialized ClientSession ready for tool calls.

Raises:

Type Description
ValueError

If config doesn't include required 'url' field.
ImportError

If mcp package is not installed.
Example

async with AsyncExitStack() as stack: ... config = {"url": "https://mcp.example.com/mcp", "type": "http"} ... session = await connect_mcp_session(stack, config) ... tools = await session.list_tools()

Source code in src/karenina/utils/mcp/client.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
async def connect_mcp_session(
    exit_stack: AsyncExitStack,
    config: MCPServerConfig,
) -> ClientSession:
    """Connect to an MCP server via HTTP/SSE transport.

    Creates a new MCP client session and registers it with the exit stack
    for automatic cleanup when the stack closes.

    Args:
        exit_stack: AsyncExitStack for managing session lifecycle.
        config: MCP server configuration. Must include 'url' for HTTP/SSE transport.
            Optional 'headers' dict for authentication. Optional 'timeout' and
            'sse_read_timeout' floats (seconds) override the MCP SDK defaults.

    Returns:
        Initialized ClientSession ready for tool calls.

    Raises:
        ValueError: If config doesn't include required 'url' field.
        ImportError: If mcp package is not installed.

    Example:
        >>> async with AsyncExitStack() as stack:
        ...     config = {"url": "https://mcp.example.com/mcp", "type": "http"}
        ...     session = await connect_mcp_session(stack, config)
        ...     tools = await session.list_tools()
    """
    from mcp import ClientSession
    from mcp.client.streamable_http import streamablehttp_client

    # Extract URL from config
    # Cast to dict for TypedDict access
    config_dict: dict[str, object] = dict(config)
    url = config_dict.get("url")
    if not url or not isinstance(url, str):
        raise ValueError("MCP server config must include 'url' for HTTP/SSE transport")

    headers_raw = config_dict.get("headers", {})
    headers: dict[str, str] = headers_raw if isinstance(headers_raw, dict) else {}

    logger.debug(f"Connecting to MCP server at {url}")

    # Create HTTP transport and enter it into the exit stack
    transport_kwargs: dict[str, Any] = {"headers": headers}
    if (timeout := config_dict.get("timeout")) is not None:
        transport_kwargs["timeout"] = timeout
    if (sse_read_timeout := config_dict.get("sse_read_timeout")) is not None:
        transport_kwargs["sse_read_timeout"] = sse_read_timeout
    http_transport = await exit_stack.enter_async_context(streamablehttp_client(url, **transport_kwargs))

    read_stream, write_stream, _ = http_transport

    # Create and initialize session
    session = await exit_stack.enter_async_context(ClientSession(read_stream, write_stream))

    await session.initialize()

    logger.debug(f"Successfully connected to MCP server at {url}")

    return session

fetch_tool_descriptions 

fetch_tool_descriptions(
    mcp_urls_dict: dict[str, str],
    tool_filter: list[str] | None = None,
) -> dict[str, str]

Synchronous wrapper for afetch_tool_descriptions.

Uses the same async handling pattern as other sync wrappers in karenina, supporting both the shared BlockingPortal (when available from parallel verification) and fallback to asyncio.run().

Parameters:

Name Type Description Default
mcp_urls_dict
dict[str, str]

Dictionary mapping server names to MCP server URLs

 required
tool_filter
list[str] | None

Optional list of tool names to include

 None

Returns:

Type Description
dict[str, str]

Dict mapping tool names to their descriptions
Source code in src/karenina/utils/mcp/tools.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def fetch_tool_descriptions(
    mcp_urls_dict: dict[str, str],
    tool_filter: list[str] | None = None,
) -> dict[str, str]:
    """Synchronous wrapper for afetch_tool_descriptions.

    Uses the same async handling pattern as other sync wrappers in karenina,
    supporting both the shared BlockingPortal (when available from parallel
    verification) and fallback to asyncio.run().

    Args:
        mcp_urls_dict: Dictionary mapping server names to MCP server URLs
        tool_filter: Optional list of tool names to include

    Returns:
        Dict mapping tool names to their descriptions
    """
    from karenina.adapters._timeouts import MCP_TOOL_FETCH_FLOOR, compute_sync_wrapper_timeout

    # No model config is available here, so the dispatch bound is the
    # historical floor for MCP tool description fetches. The inner
    # per-server connect bound (MCP_CONNECT_TIMEOUT) is derived from the
    # same source and fires first for a single server. With multiple
    # servers the per-server bounds sum, so this dispatch bound may fire
    # first instead.
    fetch_timeout = compute_sync_wrapper_timeout(None, floor=MCP_TOOL_FETCH_FLOOR)

    # Try to use the shared portal if available (from parallel verification)
    try:
        from karenina.benchmark.verification.executor import get_async_portal

        portal = get_async_portal()
        if portal is not None:
            # Bound the portal dispatch so a stalled coroutine inside the
            # portal's event loop cannot wedge the worker thread forever
            # (mirrors the langchain parser's bounded portal dispatch).
            portal_future = portal.start_task_soon(afetch_tool_descriptions, mcp_urls_dict, tool_filter)
            try:
                return portal_future.result(timeout=fetch_timeout)
            except concurrent.futures.TimeoutError as e:
                portal_future.cancel()
                raise McpTimeoutError(
                    f"Fetch tool descriptions timed out after {fetch_timeout:.0f} seconds",
                    timeout_seconds=int(fetch_timeout),
                ) from e
    except ImportError:
        pass

    # Check if we're already in an async context
    try:
        asyncio.get_running_loop()

        # Use ThreadPoolExecutor to avoid nested event loop issues
        def run_in_thread() -> dict[str, str]:
            return asyncio.run(afetch_tool_descriptions(mcp_urls_dict, tool_filter))

        with concurrent.futures.ThreadPoolExecutor() as executor:
            future = executor.submit(run_in_thread)
            try:
                return future.result(timeout=fetch_timeout)
            except TimeoutError as e:
                raise McpTimeoutError(
                    f"Fetch tool descriptions timed out after {fetch_timeout:.0f} seconds",
                    timeout_seconds=int(fetch_timeout),
                ) from e
    except RuntimeError:
        pass

    # Create new event loop and run
    return asyncio.run(afetch_tool_descriptions(mcp_urls_dict, tool_filter))

managed_mcp_server 

managed_mcp_server(
    spec: McpServerSpec,
) -> Iterator[ManagedMcpServer]

Start one local MCP server and tear it down on context exit.

Parameters:

Name Type Description Default
spec
McpServerSpec

Server command, address, timing, and logging configuration.

 required

Yields:

Type Description
ManagedMcpServer

The running server and its URL.

Raises:

Type Description
McpServerError

If the process cannot be started, become ready, or stop.
Source code in src/karenina/utils/mcp/server.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
@contextmanager
def managed_mcp_server(spec: McpServerSpec) -> Iterator[ManagedMcpServer]:
    """Start one local MCP server and tear it down on context exit.

    Args:
        spec: Server command, address, timing, and logging configuration.

    Yields:
        The running server and its URL.

    Raises:
        McpServerError: If the process cannot be started, become ready, or stop.
    """
    log_handle = _open_log(spec)
    process: subprocess.Popen[str] | None = None
    try:
        child_env = os.environ.copy()
        child_env.update(spec.env)
        try:
            process = subprocess.Popen(
                list(spec.command),
                cwd=spec.cwd,
                env=child_env,
                stdout=log_handle,
                stderr=subprocess.STDOUT,
                text=True,
                start_new_session=True,
            )
        except OSError as exc:
            raise McpServerError(
                f"Could not start MCP server {spec.name!r}: {exc}.{_diagnostic_suffix(spec)}",
                server_name=spec.name,
                log_path=str(spec.log_path) if spec.log_path else None,
            ) from exc
        try:
            _wait_until_ready(spec, process)
        except Exception:
            _stop_process(spec, process)
            raise
        logger.info("Managed MCP server %s is ready at %s", spec.name, spec.url)
        yield ManagedMcpServer(spec=spec, process=process)
    finally:
        if process is not None:
            _stop_process(spec, process)
        if not isinstance(log_handle, int):
            log_handle.close()

managed_mcp_servers 

managed_mcp_servers(
    specs: Sequence[McpServerSpec],
) -> Iterator[list[ManagedMcpServer]]

Start multiple local MCP servers and tear them down in reverse order.

Parameters:

Name Type Description Default
specs
Sequence[McpServerSpec]

Server specifications in startup order.

 required

Yields:

Type Description
list[ManagedMcpServer]

Running servers in the same order as specs.
Source code in src/karenina/utils/mcp/server.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
@contextmanager
def managed_mcp_servers(specs: Sequence[McpServerSpec]) -> Iterator[list[ManagedMcpServer]]:
    """Start multiple local MCP servers and tear them down in reverse order.

    Args:
        specs: Server specifications in startup order.

    Yields:
        Running servers in the same order as ``specs``.
    """
    with ExitStack() as stack:
        servers = [stack.enter_context(managed_mcp_server(spec)) for spec in specs]
        yield servers