JuliusBrussee/caveman · error · TypeError

Synchronous Agno calls require MiddlewareRuntime

Error message

Synchronous Agno calls require MiddlewareRuntime

What it means

The sync invoke path of the Agno middleware model requires a synchronous MiddlewareRuntime (self.connection.sync). When the adapter was configured only with an async runtime, sync model calls cannot be prepared/optimized, so a TypeError is raised before any LLM call.

Solutions

1. Provide a sync MiddlewareRuntime to the adapter/connection
2. Switch the calling code to the async path (await model.ainvoke(...))
3. Use runtime.mode 'off' or a sync-capable runtime if sync calls are required

Example fix

# before
middleware = CavemanAgno(async_runtime=AsyncMiddlewareRuntime(...))
model.invoke(messages)  # TypeError
# after
await model.ainvoke(messages)

if connection.sync is None:
    raise RuntimeError('sync invoke needs a MiddlewareRuntime; use ainvoke')

def supports_sync_invoke(model) -> bool:
    return model.connection.sync is not None

try:
    result = model.invoke(messages)
except TypeError as e:
    if 'MiddlewareRuntime' in str(e):
        result = await model.ainvoke(messages)
    else:
        raise

When it happens

Trigger: Calling model.invoke(...) on a CavemanMiddlewareModel whose connection has only an async runtime.

Common situations: An app configured with AsyncMiddlewareRuntime for async agents, but some agent/tool path calls the model synchronously; mixing sync Agno agents with an async-only middleware setup.

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 existing native Agno Model
• Match the sync/async middleware runtime to the native client
• Synchronous Agno recovery requires MiddlewareRuntime
• Agno middleware requires a nonempty native session_id

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

Appendix: source

Thrown at packages/middleware/python/caveman_middleware/agno.py:365

    def _apply_prepared(self, messages, state, outcome):
        attempt, paths, _, recheck = state
        attempt.optimization = outcome if not outcome.replacements else None
        if not recheck():
            attempt.reason = "recovery_unavailable"
            return messages
        if any(item["segment_id"] not in paths for item in outcome.replacements):
            attempt.reason = "invalid_plan"
            return messages
        view = _apply(messages, outcome, paths)
        attempt.optimization = outcome
        return view

    def invoke(self, messages, *args, **kwargs):
        self._check_cancelled()
        runtime = self.connection.sync
        if runtime is None:
            raise TypeError("Synchronous Agno calls require MiddlewareRuntime")
        state = self._prepare("invoke", messages, args, kwargs, runtime)
        attempt = None
        if state:
            attempt, _, options, _ = state
            if options is not None:
                messages = self._apply_prepared(messages, state, runtime.optimize(**options))
        self._check_cancelled()
        if attempt:
            attempt.observe("dispatch_intent")
        token = owner.set(attempt) if attempt else None
        try:
            result = self.model.invoke(messages, *args, **kwargs)
            if attempt:
                attempt.observe("completed", _usage(result))
            return result
        except BaseException:
            if attempt:
                attempt.observe("failed")

View on GitHub (pinned to 3ee70a1026)