JuliusBrussee/caveman · error · TypeError

Native MCP clients require AsyncMiddlewareRuntime

Error message

Native MCP clients require AsyncMiddlewareRuntime

What it means

CavemanMCPHost's constructor requires that the runtime passed in be an AsyncMiddlewareRuntime instance; any other object (sync runtime, None, or wrong class) raises this TypeError. Native MCP hosting is only implemented against the async runtime contract, so the guard prevents a misconfigured host from failing later at request time.

Solutions

1. Construct the runtime as AsyncMiddlewareRuntime (e.g. AsyncMiddlewareRuntime(mode=...)) and pass it to CavemanMCPHost
2. If you only have a sync MiddlewareRuntime, wrap or replace it with the async variant before building the host
3. Check that a factory/dependency-injection helper is not returning the sync runtime class

Example fix

// before
host = CavemanMCPHost(runtime=MiddlewareRuntime(mode="compress"), scope=scope, server_id="srv", protocol_version="2025-03-26")
// after
host = CavemanMCPHost(runtime=AsyncMiddlewareRuntime(mode="compress"), scope=scope, server_id="srv", protocol_version="2025-03-26")

from caveman_cloud.middleware import AsyncMiddlewareRuntime
assert isinstance(runtime, AsyncMiddlewareRuntime), "pass an AsyncMiddlewareRuntime"

def is_async_runtime(r): return isinstance(r, AsyncMiddlewareRuntime)

try:
    host = CavemanMCPHost(runtime=runtime, scope=scope, server_id=sid, protocol_version=pv)
except TypeError as e:
    if "AsyncMiddlewareRuntime" in str(e):
        runtime = AsyncMiddlewareRuntime(mode=runtime.mode)
        host = CavemanMCPHost(runtime=runtime, scope=scope, server_id=sid, protocol_version=pv)
    else:
        raise

When it happens

Trigger: Passing a sync MiddlewareRuntime, a mock/stub, or None as the `runtime` argument to CavemanMCPHost(...) instead of an AsyncMiddlewareRuntime instance.

Common situations: Developers reusing a sync runtime built for the OpenAI/Anthropic adapters, or constructing the host from a config dict where the runtime was never instantiated as the async class.

Related errors

• Use AsyncMiddlewareRuntime with the asynchronous transport
• Use MiddlewareRuntime with the synchronous transport
• ASGI context must come from authenticated server state
• Expected a native LangChain BaseChatModel
• Expected a native Strands Model

AI-assisted analysis of JuliusBrussee/caveman@3ee70a1026 (2026-09-20). Data as JSON: /api/errors/ce066b195fc36e8d. Report an issue: GitHub.

Appendix: source

Thrown at packages/middleware/python/caveman_middleware/mcp.py:44

@dataclass(frozen=True)
class MCPToolBinding:
    """The native MCP definition and the callable registered in the host loop."""
    tool: Tool
    execute: Callable[..., Awaitable[CallToolResult]]


def bind_mcp_tool(client, tool: Tool) -> MCPToolBinding:
    """Use an existing native client; it keeps auth, IDs, validation and options."""
    async def execute(arguments=None, **native_options):
        return await client.call_tool(tool.name, arguments, **native_options)
    return MCPToolBinding(tool, execute)


class CavemanMCPHost:
    def __init__(self, *, runtime: AsyncMiddlewareRuntime, scope: Scope,
                 server_id: str, protocol_version: str):
        if not isinstance(runtime, AsyncMiddlewareRuntime):
            raise TypeError("Native MCP clients require AsyncMiddlewareRuntime")
        self._version_supported = matches_framework(("mcp", "2.2", "3"))
        if not self._version_supported and runtime.mode != "off":
            runtime.decline("unsupported_version")
        if not server_id or not protocol_version:
            raise ValueError("Provide the host's server identity and negotiated protocol version")
        self.runtime, self.scope, self.server_id = runtime, scope, server_id
        self.adapter = Adapter("mcp", "0.1.0", "2.2.0", "mcp-native-" + protocol_version + "-v1")
        binding = runtime.recovery(scope)
        self._binding = binding
        tool = Tool(name=binding.name, description=binding.description, input_schema=dict(binding.input_schema))

        async def execute(arguments=None, **_native_options):
            # This is a host-local executor, not an outbound MCP tools/call.
            page = await binding.execute(arguments or {})
            return CallToolResult(content=[TextContent(type="text", text=json.dumps(page, ensure_ascii=False, separators=(",", ":")))])
        self.recovery = MCPToolBinding(tool, execute)
        self._recovery_executor = execute
        self._recovery_definition = tool.model_dump_json(by_alias=True, exclude_none=True)

View on GitHub (pinned to 3ee70a1026)