JuliusBrussee/caveman · error · TypeError

Expected an OpenAI or AsyncOpenAI client

Error message

Expected an OpenAI or AsyncOpenAI client

What it means

_wrap only accepts an official openai.OpenAI or openai.AsyncOpenAI client instance; anything else raises this TypeError. It is the entry guard for with_caveman_openai, with_caveman_openai_tools, and copy_client.

Solutions

1. Construct the client as openai.OpenAI(api_key=...) or openai.AsyncOpenAI(...) and pass that
2. If using an OpenAI-compatible provider, keep the official OpenAI client and set base_url
3. In tests, use the real client with a fake transport/base_url rather than a mock object

Example fix

// before
with_caveman_openai(httpx.Client(base_url=URL), runtime=runtime, scope=scope)
// after
with_caveman_openai(OpenAI(api_key=key, base_url=URL), runtime=runtime, scope=scope)

from openai import OpenAI
assert isinstance(client, OpenAI), "client must be openai.OpenAI or openai.AsyncOpenAI"

def is_openai_client(c):
    from openai import OpenAI, AsyncOpenAI
    return isinstance(c, (OpenAI, AsyncOpenAI))

try:
    wrapped = with_caveman_openai(client, runtime=runtime, scope=scope)
except TypeError as e:
    if "OpenAI or AsyncOpenAI" in str(e):
        client = OpenAI(api_key=key, base_url=base_url)
        wrapped = with_caveman_openai(client, runtime=runtime, scope=scope)
    else:
        raise

When it happens

Trigger: Passing a raw httpx.Client, an OpenAI subclass instance from a patched/vendored SDK, an AzureOpenAI (which is a subclass — passes only if isinstance holds), a mock, or None as `client`.

Common situations: Migrating from another wrapper library that took a base_url-configured httpx client; tests passing MagicMock clients; using a forked OpenAI-compatible SDK whose class does not inherit from openai.OpenAI.

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

• functions must map native tool names to callables
• Match the sync/async Caveman transport to the native client
• Match the sync/async middleware runtime to the native client
• ASGI context must come from authenticated server state
• Duplicate or reserved caveman_retrieve tool name

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

Appendix: source

Thrown at packages/middleware/python/caveman_middleware/openai.py:92

    if len(set(names)) != len(names) or "caveman_retrieve" in names or "caveman_retrieve" in functions:
        raise ValueError("Duplicate or reserved caveman_retrieve tool name")
    if set(names) != set(functions):
        raise ValueError("Every native function definition needs exactly one executor")
    if runtime.mode != "compress" or not in_range(__version__, "3.10", "4"):
        return CavemanOpenAIToolLoop(with_caveman_openai(client, runtime=runtime, scope=scope, transport=transport), MappingProxyType(dict(functions)), json.dumps(definitions))
    binding = runtime.recovery(scope)
    tool = {"name": binding.name, "description": binding.description, "parameters": copy.deepcopy(binding.input_schema)}
    definition = {"type": "function", "function": tool} if protocol == "openai-chat" else {"type": "function", **tool}
    definitions.append(definition)
    registry = MappingProxyType({**functions, binding.name: binding.execute})
    registration = (protocol, binding, registry, registry[binding.name], json.dumps(definition, ensure_ascii=False, separators=(",", ":")))
    return CavemanOpenAIToolLoop(_wrap(client, runtime=runtime, scope=scope, registration=registration, transport=transport), registry,
                                json.dumps(definitions, ensure_ascii=False, separators=(",", ":")))


def _wrap(client, *, runtime, scope, registration=None, transport=None):
    if not isinstance(client, (OpenAI, AsyncOpenAI)):
        raise TypeError("Expected an OpenAI or AsyncOpenAI client")
    is_async = isinstance(client, AsyncOpenAI)
    if not isinstance(runtime, AsyncMiddlewareRuntime if is_async else MiddlewareRuntime):
        raise TypeError("Match the sync/async middleware runtime to the native client")
    if transport is not None and not isinstance(transport, CavemanAsyncOpenAITransport if is_async else CavemanOpenAITransport):
        raise TypeError("Match the sync/async Caveman transport to the native client")
    version_supported = in_range(__version__, "3.10", "4")
    if not version_supported and runtime.mode != "off":
        runtime.decline("unsupported_version")
    native = client.with_options()
    post = native.post
    sessions = {}
    for path, protocol in (("/chat/completions", "openai-chat"), ("/responses", "openai-responses")):
        bound = registration is not None and registration[0] == protocol
        sessions[path] = NativeSession(runtime, scope, adapter_id="openai-sdk", framework_version="3.10.0", protocol=protocol,
                                      binding=registration[1] if bound else None, overhead=registration[4] if bound else None,
                                      is_registered=(lambda: registration[2].get(registration[1].name) is registration[3] and registration[1].execute is registration[3]) if bound else None,
                                      passive_reason=None if version_supported else "unsupported_version")
    passive_session = sessions["/chat/completions"]

View on GitHub (pinned to 3ee70a1026)