JuliusBrussee/caveman · error · TypeError

Use AsyncMiddlewareRuntime with the asynchronous transport

Error message

Use AsyncMiddlewareRuntime with the asynchronous transport

What it means

CavemanGoogleAsyncTransport requires an AsyncMiddlewareRuntime; it transparently promotes a sync MiddlewareRuntime via runtime.as_async(), but any other type fails this TypeError. It guarantees the async HTTPX transport always runs through the async bounded execution path.

Solutions

1. Pass a MiddlewareRuntime (auto-promoted via as_async()) or an AsyncMiddlewareRuntime as runtime=.
2. Ensure the runtime object is actually constructed and not None before building the transport.
3. For pure sync usage, switch to the synchronous CavemanGoogleTransport.

Example fix

// before
transport = CavemanGoogleAsyncTransport(runtime=runtime_config, scope=scope)
// after
transport = CavemanGoogleAsyncTransport(runtime=AsyncMiddlewareRuntime(...), scope=scope)

assert isinstance(runtime, (MiddlewareRuntime, AsyncMiddlewareRuntime)), 'runtime must be a Caveman runtime'
transport = CavemanGoogleAsyncTransport(runtime=runtime, scope=scope)

def is_async_compatible(r): return isinstance(r, (MiddlewareRuntime, AsyncMiddlewareRuntime))

try:
    transport = CavemanGoogleAsyncTransport(runtime=runtime, scope=scope)
except TypeError:
    transport = CavemanGoogleAsyncTransport(runtime=MiddlewareRuntime(...), scope=scope)

When it happens

Trigger: Calling CavemanGoogleAsyncTransport(runtime=X, scope=...) where X is not MiddlewareRuntime and not AsyncMiddlewareRuntime — e.g. None, a string config value, or a custom runtime-like object.

Common situations: Async app wired with a sync runtime that was already discarded; passing the runtime factory/class instead of the instance; None because the runtime failed to initialize earlier and the error was swallowed.

Related errors

• Use MiddlewareRuntime with the synchronous transport
• Match the sync/async Caveman transport to the native client
• Match the sync/async middleware runtime to the native client
• Native MCP clients require AsyncMiddlewareRuntime
• ASGI context must come from authenticated server state

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

Appendix: source

Thrown at packages/middleware/python/caveman_middleware/google.py:464

                response.stream = _SyncStream(response.stream, attempt, "text/event-stream" in response.headers.get("content-type", ""))
            return response
        except BaseException:
            attempt.observe("failed")
            raise
        finally:
            owner.reset(token)

    def close(self):
        self.transport.close()


class CavemanGoogleAsyncTransport(httpx.AsyncBaseTransport):
    """Async HTTPX transport with the runtime's bounded async execution path."""
    def __init__(self, *, runtime, scope, provider_base_url="https://generativelanguage.googleapis.com", transport=None):
        if isinstance(runtime, MiddlewareRuntime):
            runtime = runtime.as_async()
        if not isinstance(runtime, AsyncMiddlewareRuntime):
            raise TypeError("Use AsyncMiddlewareRuntime with the asynchronous transport")
        self.runtime, self.scope, self.transport = runtime, scope, transport or httpx.AsyncHTTPTransport()
        self.base = urlsplit(provider_base_url)

    async def handle_async_request(self, request):
        url = urlsplit(str(request.url))
        if (url.scheme, url.netloc) != (self.base.scheme, self.base.netloc) or not url.path.startswith(self.base.path.rstrip("/") + "/"):
            return await self.transport.handle_async_request(request)
        state = _prepare(request, self.runtime, self.scope)
        if state is None:
            return await self.transport.handle_async_request(request)
        attempt, plan = state
        outgoing = _apply(request, state, await self.runtime.optimize(**plan[2]) if plan else None)
        attempt.observe("dispatch_intent")
        token = owner.set(attempt)
        try:
            response = await self.transport.handle_async_request(outgoing)
            if response.status_code >= 400:
                attempt.observe("failed")

View on GitHub (pinned to 3ee70a1026)