JuliusBrussee/caveman · error · TypeError

ASGI requires AsyncMiddlewareRuntime

Error message

ASGI requires AsyncMiddlewareRuntime

What it means

The ASGI middleware intercepts async request/response cycles, so it only accepts an AsyncMiddlewareRuntime. Constructing it with a synchronous MiddlewareRuntime (or any other object) raises TypeError at app startup time.

Solutions

1. Construct and pass an AsyncMiddlewareRuntime to the ASGI wrapper
2. If you have a sync runtime, create a separate async runtime instance with the same configuration
3. Wrap the construction in a startup check that fails fast with a clear message

Example fix

// before
app = CavemanASGI(app, runtime=MiddlewareRuntime(...), routes=routes, resolve_context=resolve)
// after
from caveman_cloud.middleware import AsyncMiddlewareRuntime
app = CavemanASGI(app, runtime=AsyncMiddlewareRuntime(...), routes=routes, resolve_context=resolve)

from caveman_cloud.middleware import AsyncMiddlewareRuntime
if not isinstance(runtime, AsyncMiddlewareRuntime):
    raise TypeError('ASGI wrapper requires AsyncMiddlewareRuntime')

def is_async_runtime(runtime) -> bool:
    from caveman_cloud.middleware import AsyncMiddlewareRuntime
    return isinstance(runtime, AsyncMiddlewareRuntime)

try:
    asgi = CavemanASGI(app, runtime=runtime, routes=routes, resolve_context=resolve)
except TypeError as e:
    if 'AsyncMiddlewareRuntime' in str(e):
        asgi = CavemanASGI(app, runtime=AsyncMiddlewareRuntime.from_config(cfg), routes=routes, resolve_context=resolve)
    else:
        raise

When it happens

Trigger: Passing MiddlewareRuntime(...) as the `runtime` argument of the ASGI wrapper constructor, or passing None/a mock.

Common situations: Sharing one sync runtime configured for CLI/scripts into a web app; refactoring from a sync gateway to ASGI (FastAPI/Starlette) 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

• Match the sync/async middleware runtime to the native client
• Synchronous Agno calls require MiddlewareRuntime
• Synchronous Agno recovery requires MiddlewareRuntime
• ASGI context must come from authenticated server state
• ASGI context resolver or request bounds are invalid

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

Appendix: source

Thrown at packages/middleware/python/caveman_middleware/asgi.py:65

def _reject_constant(_value):
    raise ValueError("non-finite JSON")


class CavemanASGIMiddleware:
    """Place inside authentication and original-content guard middleware.

    resolve_context may decline by returning None; existing application routing
    and auth then run unchanged. It must use authenticated server state, not
    namespace/session headers. This middleware never handles an auth rejection.
    """
    def __init__(self, app, *, runtime: AsyncMiddlewareRuntime,
                 routes: Mapping[str, Protocol],
                 resolve_context: Callable[[ASGIScope], ASGIContext | None | Awaitable[ASGIContext | None]],
                 max_body_bytes: int = 2 << 20, max_request_chunks: int = 256):
        if not isinstance(runtime, AsyncMiddlewareRuntime):
            raise TypeError("ASGI requires AsyncMiddlewareRuntime")
        if not routes or any(not isinstance(path, str) or not path.startswith("/") or "*" in path or "?" in path
                             or protocol not in ("openai-chat", "openai-responses", "anthropic-messages") for path, protocol in routes.items()):
            raise ValueError("Configure exact POST paths and native LLM protocols")
        if not callable(resolve_context) or not 1 <= max_body_bytes <= 2 << 20 or not 1 <= max_request_chunks <= 4096:
            raise ValueError("ASGI context resolver or request bounds are invalid")
        self.app, self.runtime = app, runtime
        self.routes, self.resolve_context = dict(routes), resolve_context
        self.max_body_bytes, self.max_request_chunks = max_body_bytes, max_request_chunks
        self._version_supported = matches_framework(("fastapi", "0.141", "1"), ("starlette", "1.6", "2"))
        if not self._version_supported and runtime.mode != "off":
            runtime.decline("unsupported_version")

    async def __call__(self, scope, receive, send):
        if owner.get() is not None:
            return await self.app(scope, receive, send)

        protocol = self.routes.get(scope.get("path"))
        async def passthrough(reader, reason):

View on GitHub (pinned to 3ee70a1026)