Python frameworks

The thirteen published Python adapters are separate optional integrations. Start with the complete LangChain quickstart. It supplies a real local runtime, a deterministic model, tools, messages, recovery assertions, and shutdown.

Each module below is a complete integration function for configured native objects supplied by your application. It names every input explicitly and leaves framework scheduling with the framework. A build function returns a native agent; invoke it through that framework. Provider models, credentials, tools, and native HTTP clients remain your inputs. For end-to-end setup with all inputs defined, use the primary quickstart before adapting these modules.

For each family, create a fresh Python 3.13+ virtual environment and install the listed extra. Save its snippet as integration.py; run python -c "import integration" as an unpaid assembly check. Invoking a returned agent or run function against a real provider can incur charges. No provider calls occur on import.

Use MiddlewareRuntime for sync paths and AsyncMiddlewareRuntime for async-only paths. Prime ready() before calls (await async readiness). Share the client per application worker, provide trusted scope per conversation, consume/close native streams, and close runtime plus native clients at shutdown. await runtime.close() applies to the async Python client. See compatibility; extras are not guaranteed to coexist.

OpenAI#

python -m pip install "caveman-middleware[openai]==0.1.0a1"

Public module: caveman_middleware.openai.

from caveman_middleware.openai import with_caveman_openai

def run(*, client, model, messages, runtime, scope):
    runtime.ready()
    wrapped = with_caveman_openai(client, runtime=runtime, scope=scope)
    return wrapped.chat.completions.create(model=model, messages=messages)

Sync OpenAI with MiddlewareRuntime; AsyncOpenAI with AsyncMiddlewareRuntime and awaited methods. Chat Completions and Responses preserve native responses and stream objects. with_caveman_openai_tools returns client, native tools, and immutable functions for your own bounded Chat/Responses tool loop. Dispatch every function through that table and await async recovery.

Plain calls are recovery-free; Python’s supported SDK has no Chat runTools runner. Original client HTTP configuration stays native. To observe physical retry attempts, explicitly construct the public CavemanOpenAITransport / CavemanAsyncOpenAITransport exported by this module and pass that same transport via transport= to the wrapper. Without that seam, receipts describe one native operation. OpenAI 2.x can resolve under the published extra but fails the adapter gate; pin 3.10 or later below 4.

Anthropic#

python -m pip install "caveman-middleware[anthropic]==0.1.0a1"

Public module: caveman_middleware.anthropic.

from caveman_middleware.anthropic import with_caveman_anthropic

def run(*, client, model, messages, runtime, scope):
    runtime.ready()
    wrapped = with_caveman_anthropic(client, runtime=runtime, scope=scope)
    return wrapped.messages.create(model=model, max_tokens=512, messages=messages)

Match Anthropic/AsyncAnthropic to sync/async runtime. CavemanAnthropicMiddleware is the lower-level public middleware class; with_caveman_anthropic preserves middleware order and integrates native beta tool runners for recovery. Messages and streams stay native.

Plain Messages calls have no recovery executor and remain recovery-free. Existing custom HTTP clients/auth stay in the clone. Place the middleware after application authorization and original-content guards. Signed requests, extra JSON overrides, unknown shapes, and non-Messages endpoints can bypass.

Google#

python -m pip install "caveman-middleware[google]==0.1.0a1"

Public module: caveman_middleware.google.

from google import genai
from google.genai import types
from caveman_middleware.google import (
    CavemanGoogleTransport, CavemanGoogleAsyncTransport, with_caveman_google,
)

def run(*, api_key, model, contents, runtime, scope):
    runtime.ready()
    transport = CavemanGoogleTransport(runtime=runtime, scope=scope)
    async_transport = CavemanGoogleAsyncTransport(runtime=runtime.as_async(), scope=scope)
    client = genai.Client(api_key=api_key, http_options=types.HttpOptions(
        client_args={"transport": transport}, async_client_args={"transport": async_transport}))
    try:
        wrapped = with_caveman_google(client, runtime=runtime, scope=scope)
        return wrapped.models.generate_content(model=model, contents=contents)
    finally:
        client.close()

The registration helper takes a synchronous runtime and configures native sync and .aio model methods. Google automatic function calling owns execution. with_caveman_google_chat separately registers an existing Chat/AsyncChat; pass its original default config explicitly. Streams and chat history remain native.

Construct the native client with Caveman transports first; wrapping a preexisting unrelated transport alone cannot project its wire requests. Supply your original HTTPX transport through the transport constructors when needed and set its provider base URL consistently. Close both sync and async clients when you use both. Nontext, signed, unsupported content, and missing native AFC ownership bypass lossy transformations.

LangChain#

python -m pip install "caveman-middleware[langchain]==0.1.0a1"

Public module: caveman_middleware.langchain.

from langchain.agents import create_agent
from caveman_middleware.langchain import with_caveman_agent

def run(*, model, tools, messages, runtime, scope):
    runtime.ready()
    agent = create_agent(**with_caveman_agent(
        {"model": model, "tools": tools}, runtime=runtime, scope=scope))
    return agent.invoke({"messages": messages}, {"recursion_limit": 12})

CavemanMiddleware implements sync/async model middleware. with_caveman_agent registers it with its real recovery tool. with_caveman_model supports invoke/ainvoke and stream/astream without replacing LangGraph scheduling. CavemanDocumentCompressor handles explicit RAG documents.

Use sync runtime for sync calls; its async bridge also supports native async paths. scope_from_config(config, namespace=...) derives identity from trusted configurable.thread_id, branch, and epoch. Model transports stay native. RAG lossy use needs source_expansion. The complete example proves the sync tool loop with real Engine; published async stream-close/receipt gaps have unreleased fixes and are not claimed fixed here.

LiteLLM#

python -m pip install "caveman-middleware[litellm]==0.1.0a1"

Public module: caveman_middleware.litellm.

from caveman_middleware.litellm import CavemanLiteLLM

def run(*, model, messages, runtime, scope):
    runtime.ready()
    client = CavemanLiteLLM(runtime=runtime)
    try:
        return client.completion(scope=scope, model=model, messages=messages)
    finally:
        client.close()

completion, acompletion, responses, and aresponses keep LiteLLM as the inference hop. Supply client= for an existing native LiteLLM module/Router. Native streaming and callback receipt observation stay with LiteLLM.

This callback/client boundary does not invent a tool loop. Operator-owned recovery must be supplied explicitly. Proxy callbacks use a trusted proxy_scope resolver and operator_recovery; neither comes from a model header. A strict synchronous Router call is deliberately unsupported (unsupported_sync_router_strict). Preserve native provider transport configuration and close scoped registration at shutdown.

Strands#

python -m pip install "caveman-middleware[strands]==0.1.0a1"

Public module: caveman_middleware.strands.

from strands import Agent
from caveman_middleware.strands import with_caveman_agent

def build(*, model, tools, runtime, scope):
    return Agent(**with_caveman_agent(
        {"model": model, "tools": tools}, runtime=runtime, scope=scope))

CavemanModel / with_caveman_model preserve the native async model stream, token counting, and structured-output delegate. with_caveman_agent adds the native plugin and tool executor; the Strands Agent owns its synchronous facade and async scheduling. Invoke the returned agent using its native call API.

Keep the returned plugins/tools intact. Provider transport belongs to the wrapped model. Model-only use cannot attest the agent’s executor. Keep invocations and cancellation within native Strands behavior. Isolate this extra from MCP adapter versions it cannot resolve alongside.

Agno#

python -m pip install "caveman-middleware[agno]==0.1.0a1"

Public module: caveman_middleware.agno.

from agno.agent import Agent
from caveman_middleware.agno import with_caveman_agent

def build(*, model, tools, runtime, scope):
    return Agent(**with_caveman_agent(
        {"model": model, "tools": tools}, runtime=runtime, scope=scope))

CavemanModel / with_caveman_model delegate native response/invoke methods and sync/async streaming. with_caveman_agent registers the recovery function for Agno’s own tool loop. Call the returned Agent’s run or arun with your input.

scope_from_run(run, namespace=...) reads trusted run/session identity. Existing native provider models and transports remain owned by Agno. Unknown provider serialization or unsupported request shapes can bypass; do not infer support for every custom model subclass. Agno/CrewAI/LiteLLM combinations require resolver checks.

CrewAI#

python -m pip install "caveman-middleware[crewai]==0.1.0a1"

Public module: caveman_middleware.crewai.

from crewai import Agent
from caveman_middleware.crewai import with_caveman_agent

def build(*, llm, tools, runtime, scope):
    return Agent(**with_caveman_agent({
        "role": "Log analyst", "goal": "Summarize tool evidence accurately",
        "backstory": "You inspect logs and recover original evidence when needed.",
        "llm": llm, "tools": tools, "max_iter": 4,
    }, runtime=runtime, scope=scope))

CavemanLLM and with_caveman_llm wrap native call/acall; with_caveman_agent adds CavemanRecoveryTool for the actual CrewAI function registry. Your Crew and Tasks still own execution. Native streaming configuration belongs to the delegate; this is not a new standalone stream API.

Pass your configured native LLM, not a model-name string, to the wrapper. Keep task/agent context and available functions intact. Structured response modes can make recovery unavailable. Close the delegate through its owning application lifecycle. Use a separate environment from conflicting Agno or MCP dependencies.

AutoGen#

python -m pip install "caveman-middleware[autogen]==0.1.0a1"

Public module: caveman_middleware.autogen.

from autogen_agentchat.agents import AssistantAgent
from caveman_middleware.autogen import with_caveman_agent

def build(*, model_client, tools, runtime, scope):
    return AssistantAgent(**with_caveman_agent({
        "name": "log_analyst", "model_client": model_client,
        "tools": tools, "max_tool_iterations": 4,
    }, runtime=runtime, scope=scope))

CavemanChatCompletionClient / with_caveman_model expose async create and create_stream. CavemanWorkbench executes real native tools and recovery. with_caveman_agent pairs both for AssistantAgent; use native run or run_stream.

Use the native cancellation token and consume/close streams. Existing model-client transport stays native. component_runtimes supplies explicit runtime mappings when loading serialized components; secrets and runtime objects are not serialized into model config. Await wrapped model close and workbench stop as appropriate. Unsupported providers and opaque result parts bypass.

Pydantic AI#

python -m pip install "caveman-middleware[pydantic-ai]==0.1.0a1"

Public module: caveman_middleware.pydantic_ai.

from pydantic_ai import Agent
from caveman_middleware.pydantic_ai import CavemanCapability

def build(*, model, runtime, scope):
    return Agent(model, capabilities=[CavemanCapability(runtime=runtime, scope=scope)])

CavemanCapability integrates the native run context and toolset with executable recovery. CavemanModel / with_caveman_model wrap async request and request_stream; the native Agent owns run, run_sync, and streaming scheduling.

Model-only wrapping lacks complete native tool-manager registration. scope_from_run(ctx, namespace=...) uses trusted run context. Keep your provider transport on the model. Protocol recognition and exact tool-manager attestation constrain supported custom models; structured output/forced tools may remove recovery eligibility. The gate checks pydantic-ai-slim, while the installation extra is named pydantic-ai.

LlamaIndex#

python -m pip install "caveman-middleware[llama-index]==0.1.0a1"

Public module: caveman_middleware.llama_index.

from caveman_middleware.llama_index import CavemanFunctionAgent

def build(*, llm, tools, runtime, scope):
    return CavemanFunctionAgent(
        llm=llm, tools=tools, runtime=runtime, scope=scope,
        system_prompt="Use tool evidence; recover original content when needed.")

CavemanLLM / with_caveman_model delegate sync/async chat, native streams, completion, and structured methods. CavemanFunctionAgent keeps LlamaIndex workflow execution and registers recovery. with_caveman_tools supplies execute/aexecute for an application-owned tool loop. CavemanNodePostprocessor projects RAG nodes.

The installed extra includes core, not every provider integration. Recognized LLM classes are OpenAI (llama-index-llms-openai >=0.8 <1, OpenAI SDK >=2.54 <4) and Anthropic (llama-index-llms-anthropic >=0.12 <1, SDK >=0.125 <2). Other/custom classes can report unsupported_provider. Preserve node originals; lossy RAG requires source_expansion. Native provider transport and workflow Context remain application-owned.

ASGI#

python -m pip install "caveman-middleware[asgi]==0.1.0a1"

Public module: caveman_middleware.asgi.

from caveman_middleware.asgi import ASGIContext, CavemanASGIMiddleware

def attach(*, app, runtime, trusted_scope):
    async def resolve_context(asgi_scope):
        # Authentication middleware must have established this trusted server state.
        if not asgi_scope.get("state", {}).get("authenticated"):
            return None
        return ASGIContext(scope=trusted_scope(asgi_scope))
    return CavemanASGIMiddleware(app, runtime=runtime,
        routes={"/v1/chat/completions": "openai-chat"},
        resolve_context=resolve_context)

Async-only with AsyncMiddlewareRuntime. It projects bounded JSON requests on configured exact POST paths for openai-chat, openai-responses, or anthropic-messages. The original app owns routes, HTTP server, inference transport, response streaming, and shutdown.

Place inside authentication and original-content guards. The supplied trusted scope resolver must not derive authority from untrusted model or request headers. No executor is installed by this boundary: add an ASGIContext.recovery binding only when the real application loop already executes it. Without one, lossy projection is unavailable. Other routes, compressed/signed bodies, malformed JSON, excessive chunks, and unknown shapes bypass.

MCP#

python -m pip install "caveman-middleware[mcp]==0.1.0a1"

Public module: caveman_middleware.mcp.

from caveman_middleware.mcp import CavemanMCPHost, bind_mcp_tool

async def project(*, client, tool, arguments, call_id, context_manifest,
                  server_id, protocol_version, runtime, scope):
    await runtime.ready()
    host = CavemanMCPHost(runtime=runtime, scope=scope,
                         server_id=server_id, protocol_version=protocol_version)
    source = bind_mcp_tool(client, tool)
    registered = host.register([source])
    original = await source.execute(arguments)
    view = await host.project_result(original, tool=tool, call_id=call_id,
        context_manifest=context_manifest, registered_tools=registered)
    return original, view, registered

Async-only with AsyncMiddlewareRuntime. MCP tools stay native; the host owns execution, stored history, and final provider calls. Register and dispatch the returned tool bindings, including recovery, before giving the projected view to a model.

Pass your connected native ClientSession, actual protocol version, stable server ID, and original append-only manifest. bind_mcp_tool preserves native transport/auth/framing. Save original; use only view for model context. Errors, nontext blocks, and unsupported shapes are not rewritten. This does not turn an MCP server into an inference proxy.