JuliusBrussee/caveman · error · TypeError

Expected an existing native Agno Model

Error message

Expected an existing native Agno Model

What it means

CavemanMiddlewareModel is a wrapper around an existing native Agno Model instance; it copies the wrapped model's fields and delegates to it. Passing anything that is not an agno Model instance (a class, a string model id, or None) cannot be wrapped, so __init__ raises TypeError.

Solutions

1. Construct the native Agno model first and pass the instance to CavemanMiddlewareModel
2. Verify imports: Model should be the agno models base class and your instance its subclass
3. Ensure the model variable is not None due to failed conditional construction

Example fix

// before
model = CavemanMiddlewareModel("gpt-4o")
// after
from agno.models.openai import OpenAIChat
model = CavemanMiddlewareModel(OpenAIChat(id="gpt-4o"), runtime=runtime)

from agno.models.base import Model as AgnoModel
if not isinstance(candidate, AgnoModel):
    raise TypeError('pass a constructed agno Model instance')

def is_agno_model(obj) -> bool:
    from agno.models.base import Model as AgnoModel
    return isinstance(obj, AgnoModel)

try:
    wrapped = CavemanMiddlewareModel(model, runtime=runtime)
except TypeError:
    wrapped = CavemanMiddlewareModel(construct_model(model_id), runtime=runtime)

When it happens

Trigger: Passing a model identifier string (e.g. 'gpt-4o') instead of a constructed agno Model; passing the Model class rather than an instance; passing None because the model was built conditionally.

Common situations: Migrating code where a plain model name was previously accepted; confusion with providers that accept string model ids; instantiating the wrapper before creating the underlying provider model.

Related errors

• Synchronous Agno calls require MiddlewareRuntime
• Agno middleware requires a nonempty native session_id
• Agno scope resolver must return a Caveman Scope
• ASGI requires AsyncMiddlewareRuntime
• Expected a native Strands Model

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

Appendix: source

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

            overhead = json.dumps({"type": "function", "function": actual}, ensure_ascii=False, separators=(",", ":"))
        attempt = Attempt(runtime, scope, str(uuid.uuid4()), str(uuid.uuid4()), adapter=ADAPTER.id)
        recheck = lambda: binding is None or (runtime.owns_binding(binding, scope) and self.binding_details(frame, options) is not None)
        runtime_options = dict(scope=scope, adapter=ADAPTER, manifest=context, candidates=candidates, binding=binding,
            model={"provider": model.provider or type(model).__name__, "id": model.id, "protocol": "agno"}, recovery_overhead_text=overhead,
            logical_call_id=attempt.logical_call_id, attempt_id=attempt.attempt_id)
        return attempt, paths, runtime_options, recheck


class CavemanModel(Model):
    """Native Agno Model delegate. Model-only use has no recovery executor.

    Agno's Model.response methods retain orchestration. Only invoke/ainvoke and
    their streaming equivalents receive a separate model-facing Message view.
    The caller owns both its existing provider clients and the shared runtime.
    """
    def __init__(self, model: Model, *, runtime=None, scope=None, connection=None):
        if not isinstance(model, Model):
            raise TypeError("Expected an existing native Agno Model")
        super().__init__(**{item.name: getattr(model, item.name) for item in fields(Model) if not item.name.startswith("_")})
        self.model = model
        self.connection = connection or _Connection(runtime, scope)
        self.signatures = {name: inspect.signature(getattr(model, name)) for name in ("invoke", "ainvoke", "invoke_stream", "ainvoke_stream")}

    def get_provider(self):
        return self.model.get_provider()

    def to_dict(self):
        return self.model.to_dict()

    def count_tokens(self, *args, **kwargs):
        return self.model.count_tokens(*args, **kwargs)

    async def acount_tokens(self, *args, **kwargs):
        return await self.model.acount_tokens(*args, **kwargs)

    def get_system_message_for_model(self, tools=None):

View on GitHub (pinned to 3ee70a1026)