JuliusBrussee/caveman · error · TypeError

Synchronous LiteLLM calls require MiddlewareRuntime

Error message

Synchronous LiteLLM calls require MiddlewareRuntime

What it means

Synchronous LiteLLM calls (completion, responses) execute on the caller's thread and need the full MiddlewareRuntime surface (strict mode, sync recovery, reporting). _sync checks isinstance(self.runtime, MiddlewareRuntime) and raises TypeError if a lighter/async-only runtime object was supplied.

Solutions

  1. Construct the adapter with a full MiddlewareRuntime instance for sync usage
  2. Create a separate adapter with a sync-capable runtime instead of reusing the async-configured one
  3. If you are in async code, call acompletion/aresponses instead of the sync methods
  4. In tests, use the real MiddlewareRuntime (or a subclass) rather than arbitrary mocks

Example fix

// before
async_runtime = runtime.as_async()
adapter = CavemanLiteLLM(runtime=async_runtime)
adapter.completion(scope=scope, ...)  # TypeError

// after
adapter = CavemanLiteLLM(runtime=runtime)
adapter.completion(scope=scope, ...)
Defensive patterns

Strategy: type-guard

Validate before calling

from caveman_cloud.middleware import MiddlewareRuntime
if not isinstance(runtime, MiddlewareRuntime):
    raise TypeError("sync LiteLLM calls require a MiddlewareRuntime instance")

Type guard

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

Try / catch

try:
    return adapter.completion(scope=scope, **kwargs)
except TypeError as e:
    if "require MiddlewareRuntime" in str(e):
        adapter = CavemanLiteLLM(runtime=middleware_runtime, client=adapter.client)
        return adapter.completion(scope=scope, **kwargs)

Prevention

When it happens

Trigger: Constructing CavemanLiteLLM with an async-only runtime (e.g. the result of runtime.as_async() or a custom object) and then invoking the sync completion()/responses() path.

Common situations: Sharing one adapter configured for an async app (FastAPI) inside sync scripts or Celery workers; passing a mock runtime in tests that isn't a MiddlewareRuntime subclass.

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


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

Appendix: source

Thrown at packages/middleware/python/caveman_middleware/litellm.py:302

            owner.reset(owner_token)
            _active.reset(active_token)

    def _sync(self, method, scope, kwargs):
        function = getattr(self.client, method)
        if owner.get() is not None:
            return function(**kwargs)
        passive_reason = self._passive_reason(method, kwargs)
        if passive_reason:
            attempt = Attempt(self.runtime, None, str(uuid.uuid4()), str(uuid.uuid4()), passive=True,
                              reason=passive_reason, adapter="litellm")
            attempt.observe("dispatch_intent")
            token = owner.set(attempt)
            try:
                return function(**kwargs)
            finally:
                owner.reset(token)
        if not isinstance(self.runtime, MiddlewareRuntime):
            raise TypeError("Synchronous LiteLLM calls require MiddlewareRuntime")
        router = isinstance(self.client, native.Router) and method == "completion"
        if router and self.runtime.strict:
            # Logging callbacks swallow their exceptions. Reject this capability
            # before calling the Router so strict failure cannot dispatch a request.
            self._report("unsupported_sync_router_strict")
            raise MiddlewareError("unsupported_sync_router_strict")
        if router and self._registrations == 0:
            self._report("not_registered")
            raise RuntimeError("Use CavemanLiteLLM as a context manager while sync Router calls and streams are active")
        with self._activation(scope, kwargs, method) as (params, request):
            if router:
                token = _sync_router.set(self)
                try:
                    return function(**params)
                finally:
                    _sync_router.reset(token)
            # Direct sync SDK calls already name the selected provider/model.
            session = self._session(request, method, self.runtime) if self.client is native else None

View on GitHub (pinned to 3ee70a1026)