JuliusBrussee/caveman · error · TypeError

Expected an AutoGen Workbench or list of workbenches

Error message

Expected an AutoGen Workbench or list of workbenches

What it means

CavemanWorkbench wraps one or more AutoGen Workbench instances to add Caveman recovery tooling. The constructor validates every delegate is a Workbench instance and raises this TypeError otherwise, because it forwards tool calls and state (save_state/load_state) to each delegate.

Solutions

1. Pass an AutoGen Workbench instance, or a list whose every element is a Workbench.
2. If starting from tools, first construct a Workbench (e.g. via the AutoGen MCP or tool workbench adapters) that supports the same autogen-core version as the middleware.
3. If you have serialized configs, load each with Workbench.load_component(...) before passing.
4. Check imports: the isinstance check is against AutoGen's Workbench base class, so subclass/mix from that.

Example fix

// before
wb = CavemanWorkbench([tool_a, tool_b], runtime=rt, scope=scope)
// after
from autogen_agentchat.tools import Workbench
wb = CavemanWorkbench([MyToolWorkbench(tool_a), MyToolWorkbench(tool_b)], runtime=rt, scope=scope)

delegates = wb if isinstance(wb, list) else [wb]
if not all(isinstance(d, Workbench) for d in delegates):
    raise TypeError("each delegate must be an AutoGen Workbench")

def is_workbench_list(obj) -> bool:
    items = obj if isinstance(obj, list) else [obj]
    return bool(items) and all(isinstance(i, Workbench) for i in items)

try:
    wrapper = CavemanWorkbench(wb, runtime=rt, scope=scope)
except TypeError as e:
    if "Workbench" in str(e):
        wrapper = CavemanWorkbench(load_component(wb), runtime=rt, scope=scope)
    else:
        raise

When it happens

Trigger: Passing a single non-Workbench object (MCP adapter, tool list, None, string) or a list containing any non-Workbench item as the workbench argument to CavemanWorkbench(...).

Common situations: Passing an MCPWorkbench-like object from another library or version; passing a plain list of tools/functions instead of Workbench objects; passing a config dict loaded from JSON instead of constructed Workbench instances.

Related errors

• Expected an AutoGen ChatCompletionClient
• AutoGen requires a stable Caveman Scope for each agent or…
• Expected a native Strands Model
• Expected an existing native LlamaIndex LLM
• Expected an existing native Pydantic AI Model

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

Appendix: source

Thrown at packages/middleware/python/caveman_middleware/autogen.py:379

                attempt.observe("failed")
            raise
        finally:
            await iterator.aclose()


class CavemanWorkbench(StaticStreamWorkbench):
    """Add scoped recovery to a real workbench without changing its tool results.

    AutoGen 0.7.5 selects streaming tools by this public base class. Delegating
    call_tool_stream preserves native streaming workbenches and their events.
    """
    component_provider_override = "caveman_middleware.autogen.CavemanWorkbench"
    component_config_schema = CavemanWorkbenchConfig

    def __init__(self, workbench: Workbench | list[Workbench], *, runtime, scope: Scope, runtime_key="default"):
        delegates = workbench if type(workbench) is list else [workbench]
        if not all(isinstance(item, Workbench) for item in delegates):
            raise TypeError("Expected an AutoGen Workbench or list of workbenches")
        super().__init__([])
        self.workbench = workbench
        self.delegates = tuple(delegates)
        self.runtime, self.scope = _check(runtime, scope), scope
        self.runtime_key = runtime_key
        self.binding = self.runtime.recovery(scope)
        self.recovery_schema = _RecoverySchema(self)
        self.recovery_enabled = False
        self.stopped = False

    async def list_tools(self):
        tools = [tool for workbench in self.delegates for tool in await workbench.list_tools()]
        self.recovery_enabled = (not self.stopped and _supported(self.runtime) and self.runtime.mode == "compress"
                                 and not any(t.get("name") == "caveman_retrieve" for t in tools))
        return [*tools, self.recovery_schema] if self.recovery_enabled else tools

    async def call_tool(self, name, arguments=None, cancellation_token=None, call_id=None):
        # Refresh dynamic registries before deciding which implementation owns

View on GitHub (pinned to 3ee70a1026)