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

The middleware runtime's sync/async flavor must match the client: AsyncOpenAI requires AsyncMiddlewareRuntime, sync OpenAI requires MiddlewareRuntime. Mismatched pairings raise this TypeError because the wrapper hooks sync vs async transports differently.

Solutions

1. Pair AsyncOpenAI with AsyncMiddlewareRuntime and sync OpenAI with MiddlewareRuntime
2. Create a separate async runtime for the async client instead of reusing the sync one
3. Check factory helpers return the matching variant for your client type

Example fix

// before
with_caveman_openai(AsyncOpenAI(api_key=key), runtime=MiddlewareRuntime(mode="compress"), scope=scope)
// after
with_caveman_openai(AsyncOpenAI(api_key=key), runtime=AsyncMiddlewareRuntime(mode="compress"), scope=scope)

from openai import OpenAI, AsyncOpenAI
from caveman_cloud.middleware import MiddlewareRuntime, AsyncMiddlewareRuntime
if isinstance(client, AsyncOpenAI):
    assert isinstance(runtime, AsyncMiddlewareRuntime)
else:
    assert isinstance(runtime, MiddlewareRuntime)

def runtime_matches_client(client, runtime):
    from openai import OpenAI, AsyncOpenAI
    from caveman_cloud.middleware import MiddlewareRuntime, AsyncMiddlewareRuntime
    return isinstance(runtime, AsyncMiddlewareRuntime if isinstance(client, AsyncOpenAI) else MiddlewareRuntime)

try:
    wrapped = with_caveman_openai(client, runtime=runtime, scope=scope)
except TypeError as e:
    if "sync/async middleware runtime" in str(e):
        runtime = AsyncMiddlewareRuntime(mode=runtime.mode) if isinstance(client, AsyncOpenAI) else MiddlewareRuntime(mode=runtime.mode)
        wrapped = with_caveman_openai(client, runtime=runtime, scope=scope)
    else:
        raise

When it happens

Trigger: Calling with_caveman_openai(AsyncOpenAI(...), runtime=MiddlewareRuntime(...)) or with_caveman_openai(OpenAI(...), runtime=AsyncMiddlewareRuntime(...)).

Common situations: Sharing one runtime between sync scripts and async apps; refactoring an app to asyncio and swapping the client but not the runtime; copy-pasted examples mixing variants.

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

• Match the sync/async Caveman transport to the native client
• Expected an OpenAI or AsyncOpenAI client
• functions must map native tool names to callables
• Use AsyncMiddlewareRuntime with the asynchronous transport
• ASGI context must come from authenticated server state

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

Appendix: source

Thrown at packages/middleware/python/caveman_middleware/openai.py:95

        raise ValueError("Every native function definition needs exactly one executor")
    if runtime.mode != "compress" or not in_range(__version__, "3.10", "4"):
        return CavemanOpenAIToolLoop(with_caveman_openai(client, runtime=runtime, scope=scope, transport=transport), MappingProxyType(dict(functions)), json.dumps(definitions))
    binding = runtime.recovery(scope)
    tool = {"name": binding.name, "description": binding.description, "parameters": copy.deepcopy(binding.input_schema)}
    definition = {"type": "function", "function": tool} if protocol == "openai-chat" else {"type": "function", **tool}
    definitions.append(definition)
    registry = MappingProxyType({**functions, binding.name: binding.execute})
    registration = (protocol, binding, registry, registry[binding.name], json.dumps(definition, ensure_ascii=False, separators=(",", ":")))
    return CavemanOpenAIToolLoop(_wrap(client, runtime=runtime, scope=scope, registration=registration, transport=transport), registry,
                                json.dumps(definitions, ensure_ascii=False, separators=(",", ":")))


def _wrap(client, *, runtime, scope, registration=None, transport=None):
    if not isinstance(client, (OpenAI, AsyncOpenAI)):
        raise TypeError("Expected an OpenAI or AsyncOpenAI client")
    is_async = isinstance(client, AsyncOpenAI)
    if not isinstance(runtime, AsyncMiddlewareRuntime if is_async else MiddlewareRuntime):
        raise TypeError("Match the sync/async middleware runtime to the native client")
    if transport is not None and not isinstance(transport, CavemanAsyncOpenAITransport if is_async else CavemanOpenAITransport):
        raise TypeError("Match the sync/async Caveman transport to the native client")
    version_supported = in_range(__version__, "3.10", "4")
    if not version_supported and runtime.mode != "off":
        runtime.decline("unsupported_version")
    native = client.with_options()
    post = native.post
    sessions = {}
    for path, protocol in (("/chat/completions", "openai-chat"), ("/responses", "openai-responses")):
        bound = registration is not None and registration[0] == protocol
        sessions[path] = NativeSession(runtime, scope, adapter_id="openai-sdk", framework_version="3.10.0", protocol=protocol,
                                      binding=registration[1] if bound else None, overhead=registration[4] if bound else None,
                                      is_registered=(lambda: registration[2].get(registration[1].name) is registration[3] and registration[1].execute is registration[3]) if bound else None,
                                      passive_reason=None if version_supported else "unsupported_version")
    passive_session = sessions["/chat/completions"]

    def session_for(path, kwargs):
        if not isinstance(path, str) or not plain(kwargs.get("options", {})) or kwargs.get("options", {}).get("extra_json") or kwargs.get("files"):

View on GitHub (pinned to 3ee70a1026)