JuliusBrussee/caveman · error · TypeError

Match the sync/async middleware runtime to the native client

Error message

Match the sync/async middleware runtime to the native client

What it means

After identifying the client as sync (Anthropic) or async (AsyncAnthropic), with_caveman_anthropic checks that the runtime kind matches: AsyncMiddlewareRuntime for async clients, MiddlewareRuntime for sync clients. Mismatched kinds would produce broken call semantics, so TypeError is raised.

Solutions

1. Match the runtime class to the client: MiddlewareRuntime for Anthropic, AsyncMiddlewareRuntime for AsyncAnthropic
2. Construct both a sync and async runtime if you serve both client kinds
3. Detect at the call site with isinstance(client, AsyncAnthropic) and pick the runtime accordingly

Example fix

// before
with_caveman_anthropic(AsyncAnthropic(), runtime=MiddlewareRuntime(...))  # TypeError
// after
from caveman_cloud.middleware import AsyncMiddlewareRuntime
with_caveman_anthropic(AsyncAnthropic(), runtime=AsyncMiddlewareRuntime(...), scope=scope)

from anthropic import AsyncAnthropic
from caveman_cloud.middleware import MiddlewareRuntime, AsyncMiddlewareRuntime
expected = AsyncMiddlewareRuntime if isinstance(client, AsyncAnthropic) else MiddlewareRuntime
if not isinstance(runtime, expected):
    raise TypeError('runtime kind must match client kind')

def runtime_matches(client, runtime) -> bool:
    from anthropic import AsyncAnthropic
    from caveman_cloud.middleware import MiddlewareRuntime, AsyncMiddlewareRuntime
    return isinstance(runtime, AsyncMiddlewareRuntime if isinstance(client, AsyncAnthropic) else MiddlewareRuntime)

try:
    wrapped = with_caveman_anthropic(client, runtime=runtime, scope=scope)
except TypeError as e:
    if 'Match the sync/async' in str(e):
        wrapped = with_caveman_anthropic(client, runtime=rebuild_runtime_for(client), scope=scope)
    else:
        raise

When it happens

Trigger: Calling with_caveman_anthropic(AsyncAnthropic(), runtime=MiddlewareRuntime(...)) or with_caveman_anthropic(Anthropic(), runtime=AsyncMiddlewareRuntime(...)).

Common situations: Reusing one shared runtime instance for both sync and async clients; copying configuration between a sync script and async service without swapping the runtime class.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• ASGI requires AsyncMiddlewareRuntime
• Expected an Anthropic or AsyncAnthropic client
• Synchronous Agno calls require MiddlewareRuntime
• Synchronous Agno recovery requires MiddlewareRuntime
• Expected an existing native Agno Model

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

Appendix: source

Thrown at packages/middleware/python/caveman_middleware/anthropic.py:117

class _AsyncRecoveryTool(BetaAsyncBuiltinFunctionTool):
    binding: object
    definition: str
    execute: object

    def to_dict(self):
        return json.loads(self.definition)

    async def call(self, input):
        return json.dumps(await self.execute(input), ensure_ascii=False, separators=(",", ":"))


def with_caveman_anthropic(client, *, runtime, scope):
    """Return a native SDK clone. Existing native middleware stays in order."""
    if not isinstance(client, (Anthropic, AsyncAnthropic)):
        raise TypeError("Expected an Anthropic or AsyncAnthropic client")
    async_client = isinstance(client, AsyncAnthropic)
    if not isinstance(runtime, AsyncMiddlewareRuntime if async_client else MiddlewareRuntime):
        raise TypeError("Match the sync/async middleware runtime to the native client")
    version_supported = in_range(__version__, "1.4", "2")
    if not version_supported and runtime.mode != "off":
        runtime.decline("unsupported_version")
    native = client.with_middleware(CavemanAnthropicMiddleware(runtime, scope))
    if not version_supported or runtime.mode == "off":
        return native
    original_runner = native.beta.messages.tool_runner

    @functools.wraps(original_runner)
    def tool_runner(**params):
        supplied = params.get("tools")
        if runtime.mode != "compress" or not isinstance(supplied, (list, tuple)):
            return original_runner(**params)
        names = [(t.get("name") if plain(t) else getattr(t, "name", None)) for t in supplied]
        if any(type(name) is not str or not name for name in names) or len(set(names)) != len(names) or "caveman_retrieve" in names:
            return original_runner(**params)
        binding = runtime.recovery(scope)
        definition = json.dumps({"name": binding.name, "description": binding.description, "input_schema": binding.input_schema}, ensure_ascii=False, separators=(",", ":"))

View on GitHub (pinned to 3ee70a1026)