data_designer.engine.mcp.io

Low-level MCP I/O operations with caching and session pooling.

This module provides stateless functions for MCP communication using an actor-style service that owns all async state within a single background event loop. Public APIs are synchronous wrappers that submit coroutines to the loop and wait for results.

Architecture: All MCP I/O is funneled through a single dedicated asyncio event loop running in a background daemon thread. This avoids the complexity of managing multiple event loops and allows sessions to be reused across calls from any thread.

Worker Thread 1 ──┐ Worker Thread 2 ──┼──► MCP Event Loop Thread ──► MCP Servers Worker Thread N ──┘ (all sessions live here)

Request Coalescing: When multiple threads request tools from the same provider simultaneously, only one request is made to the MCP server. Other callers wait for the in-flight request to complete and share the result. This prevents N concurrent workers from making N separate ListToolsRequest calls.

The caller (MCPFacade) is responsible for resolving any secret references in provider api_key fields before passing providers to these functions.

Module Contents

Classes

Name	Description
`MCPIOService`	Actor-style MCP I/O service owning all async state.

Functions

Name	Description
`_provider_cache_key`	Create a stable cache key for a provider.
`list_tools`	List tools from an MCP provider (cached with request coalescing).
`list_tool_names`	Return the names of all tools available on an MCP provider.
`call_tools`	Call multiple tools in parallel.
`clear_provider_caches`	Clear all caches for specific MCP providers.
`clear_tools_cache`	Clear the list_tools cache.
`get_cache_info`	Get cache statistics for list_tools.
`clear_session_pool`	Clear all pooled MCP sessions.
`get_session_pool_info`	Get information about the session pool.
`_build_auth_headers`	Build authentication headers for remote MCP clients.
`_coerce_tool_definition`	Coerce a tool from various formats into MCPToolDefinition.
`_serialize_tool_result_content`	Serialize tool result content to a string.

Data

logger _MCP_IO_SERVICE

API

1 logger = getLogger(...)

1 data_designer.engine.mcp.io._provider_cache_key(provider: data_designer.config.mcp.MCPProviderT) -> str

Create a stable cache key for a provider.

1 class data_designer.engine.mcp.io.MCPIOService

Actor-style MCP I/O service owning all async state.

1 list_tools(
2     provider: data_designer.config.mcp.MCPProviderT,
3     timeout_sec: float | None = None
4 ) -> tuple[data_designer.engine.mcp.registry.MCPToolDefinition, ...]

List tools from an MCP provider (cached with request coalescing).

1 call_tools(
2     calls: list[tuple[data_designer.config.mcp.MCPProviderT, str, dict[str, typing.Any]]],
3     *,
4     timeout_sec: float | None = None
5 ) -> list[data_designer.engine.mcp.registry.MCPToolResult]

Call multiple tools in parallel.

1 clear_provider_caches(providers: list[data_designer.config.mcp.MCPProviderT]) -> int

Clear caches and session pool entries for specific providers.

1 clear_tools_cache() -> None

Clear the list_tools cache (best effort).

1 get_cache_info() -> dict[str, typing.Any]

Get cache statistics for list_tools.

1 clear_session_pool() -> None

Clear all pooled MCP sessions (best effort).

1 get_session_pool_info() -> dict[str, typing.Any]

Get information about the session pool.

1 shutdown() -> None

Shutdown the MCP event loop and close all sessions.

1 _ensure_loop() -> asyncio.AbstractEventLoop

1 _run_loop(loop: asyncio.AbstractEventLoop) -> None

1 _run_on_loop(
2     coro: collections.abc.Coroutine[typing.Any, typing.Any, typing.Any],
3     timeout_sec: float | None
4 ) -> typing.Any

1 _get_or_create_session(provider: data_designer.config.mcp.MCPProviderT) -> mcp.ClientSession

1 _list_tools_async(provider: data_designer.config.mcp.MCPProviderT) -> tuple[data_designer.engine.mcp.registry.MCPToolDefinition, ...]

1 _call_tool_async(
2     provider: data_designer.config.mcp.MCPProviderT,
3     tool_name: str,
4     arguments: dict[str, typing.Any]
5 ) -> data_designer.engine.mcp.registry.MCPToolResult

1 _call_tools_async(calls: list[tuple[data_designer.config.mcp.MCPProviderT, str, dict[str, typing.Any]]]) -> list[data_designer.engine.mcp.registry.MCPToolResult]

1 _clear_provider_caches_async(providers: list[data_designer.config.mcp.MCPProviderT]) -> int

1 _clear_provider_caches_sync(providers: list[data_designer.config.mcp.MCPProviderT]) -> int

1 _clear_tools_cache_async() -> None

1 _clear_tools_cache_sync() -> None

1 _get_cache_info_async() -> dict[str, typing.Any]

1 _close_all_sessions_async() -> None

1 _clear_session_pool_sync() -> None

1 _get_session_pool_info_async() -> dict[str, typing.Any]

1 _invalidate_tools_cache(keys: collections.abc.Iterable[str]) -> None

1 _all_tools_keys() -> set[str]

1 _reset_state() -> None

1 _MCP_IO_SERVICE = MCPIOService(...)

1 data_designer.engine.mcp.io.list_tools(
2     provider: data_designer.config.mcp.MCPProviderT,
3     timeout_sec: float | None = None
4 ) -> tuple[data_designer.engine.mcp.registry.MCPToolDefinition, ...]

List tools from an MCP provider (cached with request coalescing).

1 data_designer.engine.mcp.io.list_tool_names(
2     provider: data_designer.config.mcp.MCPProviderT,
3     timeout_sec: float
4 ) -> list[str]

Return the names of all tools available on an MCP provider.

1 data_designer.engine.mcp.io.call_tools(
2     calls: list[tuple[data_designer.config.mcp.MCPProviderT, str, dict[str, typing.Any]]],
3     *,
4     timeout_sec: float | None = None
5 ) -> list[data_designer.engine.mcp.registry.MCPToolResult]

Call multiple tools in parallel.

1 data_designer.engine.mcp.io.clear_provider_caches(providers: list[data_designer.config.mcp.MCPProviderT]) -> int

Clear all caches for specific MCP providers.

1 data_designer.engine.mcp.io.clear_tools_cache() -> None

Clear the list_tools cache.

1 data_designer.engine.mcp.io.get_cache_info() -> dict[str, typing.Any]

Get cache statistics for list_tools.

1 data_designer.engine.mcp.io.clear_session_pool() -> None

Clear all pooled MCP sessions.

1 data_designer.engine.mcp.io.get_session_pool_info() -> dict[str, typing.Any]

Get information about the session pool.

1 data_designer.engine.mcp.io._build_auth_headers(api_key: str | None) -> dict[str, typing.Any] | None

Build authentication headers for remote MCP clients.

1 data_designer.engine.mcp.io._coerce_tool_definition(
2     tool: typing.Any,
3     tool_definition_cls: type[data_designer.engine.mcp.registry.MCPToolDefinition]
4 ) -> data_designer.engine.mcp.registry.MCPToolDefinition

Coerce a tool from various formats into MCPToolDefinition.

1 data_designer.engine.mcp.io._serialize_tool_result_content(result: typing.Any) -> str

Serialize tool result content to a string.

1	list_tools(
2	provider: data_designer.config.mcp.MCPProviderT,
3	timeout_sec: float \| None = None
4	) -> tuple[data_designer.engine.mcp.registry.MCPToolDefinition, ...]

1	call_tools(
2	calls: list[tuple[data_designer.config.mcp.MCPProviderT, str, dict[str, typing.Any]]],
3	*,
4	timeout_sec: float \| None = None
5	) -> list[data_designer.engine.mcp.registry.MCPToolResult]

1	_run_on_loop(
2	coro: collections.abc.Coroutine[typing.Any, typing.Any, typing.Any],
3	timeout_sec: float \| None
4	) -> typing.Any

1	_call_tool_async(
2	provider: data_designer.config.mcp.MCPProviderT,
3	tool_name: str,
4	arguments: dict[str, typing.Any]
5	) -> data_designer.engine.mcp.registry.MCPToolResult

1	data_designer.engine.mcp.io.list_tools(
2	provider: data_designer.config.mcp.MCPProviderT,
3	timeout_sec: float \| None = None
4	) -> tuple[data_designer.engine.mcp.registry.MCPToolDefinition, ...]

1	data_designer.engine.mcp.io.list_tool_names(
2	provider: data_designer.config.mcp.MCPProviderT,
3	timeout_sec: float
4	) -> list[str]

1	data_designer.engine.mcp.io.call_tools(
2	calls: list[tuple[data_designer.config.mcp.MCPProviderT, str, dict[str, typing.Any]]],
3	*,
4	timeout_sec: float \| None = None
5	) -> list[data_designer.engine.mcp.registry.MCPToolResult]

1	data_designer.engine.mcp.io._coerce_tool_definition(
2	tool: typing.Any,
3	tool_definition_cls: type[data_designer.engine.mcp.registry.MCPToolDefinition]
4	) -> data_designer.engine.mcp.registry.MCPToolDefinition