--- title: LangChain and Python summary: The Python runtime with LangChain, and the same pattern for the other Python frameworks. canonical: https://docs.caveman.so/docs/sdk/middleware/python license: MIT capability: sdk-python updated: 2026-09-16T21:32:40-07:00 basis: inferred --- # LangChain and Python > The Python runtime with LangChain, and the same pattern for the other Python frameworks. `caveman_middleware` holds one adapter per Python framework. Each takes the same `runtime` and `scope` pair, swaps eligible tool results for a shorter copy before the provider call, and registers `caveman_retrieve` so the model can read the original back. The runtime client itself ships inside `caveman-sdk` as `caveman_cloud.middleware`. ## Install Both are on PyPI. `caveman-middleware` has only prereleases so far, so name the version or pass `--pre`. Each framework family is an extra. ```bash pip install caveman-sdk "caveman-middleware[langchain]==0.1.0a1" ``` The extras are `openai`, `anthropic`, `google`, `langchain`, `litellm`, `strands`, `agno`, `asgi`, `mcp`, `crewai`, `pydantic-ai`, `autogen` and `llama-index`. Python 3.13 or newer. ## LangChain `with_caveman_agent` rewrites the keyword arguments you were going to pass to `create_agent`. It returns a new dict, so the one you built is untouched. ```python 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}) ``` `with_caveman_model` wraps a single `BaseChatModel` instead, for graphs that construct the agent themselves. `scope_from_config(config, namespace=...)` builds a `Scope` from a LangGraph `RunnableConfig` so one thread keeps one identity across turns. ## Build the runtime `Scope` is a dataclass of four strings. They key the stored originals, so a handle from an earlier turn only resolves under the same four. ```python from caveman_cloud.middleware import MiddlewareRuntime, Scope runtime = MiddlewareRuntime(endpoint="http://127.0.0.1:8787", mode="compress", deadline_ms=500) scope = Scope("support", conversation_id, "main", "0") ``` `endpoint` defaults to `http://127.0.0.1:8787`, `mode` to `compress`, `deadline_ms` to 100 and `retrieve_deadline_ms` to 5000. `runtime.as_async()` returns an async view over the same connection for `await`-based frameworks. `runtime.close()` releases both. ## Verify This is the program the documentation runs against a real runtime. It scripts the model, so it makes no provider call and costs nothing. `DEMO_MODE` selects `off`, `record` or `compress`. [Download quickstart.py](/examples/middleware/quickstart.py) ```python import copy import hashlib import json import os import re import time import uuid from caveman_cloud.middleware import MiddlewareRuntime, Scope from caveman_middleware.langchain import with_caveman_agent from langchain.agents import create_agent from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages import AIMessage, HumanMessage, ToolMessage from langchain_core.outputs import ChatGeneration, ChatResult from langchain_core.tools import tool mode = os.getenv("DEMO_MODE", "record") if mode not in ("off", "record", "compress"): raise ValueError("Invalid DEMO_MODE") paid = os.getenv("PAID_PROVIDER") == "1" original = "".join(f"[INFO] worker=alpha request={i} status=healthy latency_ms=12 café 🌍\r\n" for i in range(800)) reports = [] def on_report(report): from dataclasses import asdict reports.append(report) print(json.dumps(asdict(report))) runtime = MiddlewareRuntime(endpoint=os.getenv("CAVEMAN_ENDPOINT", "http://127.0.0.1:8787"), token=os.getenv("CAVEMAN_AUTH_TOKEN"), mode=mode, deadline_ms=500, retrieve_deadline_ms=5000, on_report=on_report) # In an application, derive namespace/session from authenticated account/conversation IDs. scope = Scope("docs-demo", str(uuid.uuid4()), "main", "1") @tool def read_log() -> str: """Read the demo worker log.""" return original history = [HumanMessage("Read log; recover all original pages if shortened, then summarize status."), AIMessage("", tool_calls=[{"id": "read-1", "name": "read_log", "args": {}}]), ToolMessage(original, tool_call_id="read-1", name="read_log")] for i, message in enumerate(history): message.id = f"history-{i}" before = copy.deepcopy(history) class ScriptedModel(BaseChatModel): handle: str | None = None recovery_calls: int = 0 @property def _llm_type(self): return "deterministic-caveman-docs-fixture" def bind_tools(self, tools, **kwargs): return self def _generate(self, messages, stop=None, run_manager=None, **kwargs): match = re.search(r"cmw_[a-f0-9]{48}", str(messages)) if match: self.handle = match.group(0) recovered = any(isinstance(m, ToolMessage) and m.name == "caveman_retrieve" for m in messages) if self.handle and not recovered: self.recovery_calls += 1 output = AIMessage("", tool_calls=[{"id": "recover-1", "name": "caveman_retrieve", "args": {"handle": self.handle, "limit": 262144}}]) else: output = AIMessage("Deterministic fixture completed.") return ChatResult(generations=[ChatGeneration(message=output)]) try: if mode != "off": caps = runtime.ready() print("capabilities", {key: caps[key] for key in ("mode", "persistent", "recovery", "retention_seconds")}) model = ScriptedModel() if paid: if not os.getenv("OPENAI_API_KEY") or not os.getenv("OPENAI_MODEL"): raise ValueError("Set OPENAI_API_KEY and OPENAI_MODEL for paid run") from langchain_openai import ChatOpenAI model = ChatOpenAI(model=os.environ["OPENAI_MODEL"], max_retries=0) agent = create_agent(**with_caveman_agent({"model": model, "tools": [read_log]}, runtime=runtime, scope=scope)) started = time.perf_counter() result = agent.invoke({"messages": history}, {"recursion_limit": 12}) assert history == before, "Application history must retain original bytes" if not paid and mode == "off": assert any(r.status == "disabled" for r in reports) if not paid and mode == "record": assert any(r.status == "recorded" for r in reports) if not paid and mode == "compress": assert any(r.status in ("applied", "reused") for r in reports), "Expected compression; inspect reports" assert model.recovery_calls == 1, "Native loop must execute recovery" offset, restored = 0, "" while offset is not None: page = runtime.retrieve(scope, handle=model.handle, offset=offset, limit=4096) assert page["kind"] == "original_page" restored += page["text"] offset = page["next_offset"] assert restored == original print("exact recovery SHA-256", hashlib.sha256(restored.encode()).hexdigest()) print({"text": result["messages"][-1].content, "elapsed_ms": round((time.perf_counter()-started)*1000), "usage": [m.usage_metadata for m in result["messages"] if isinstance(m, AIMessage)] if paid else "fixture; not provider usage", "unchanged_history": True}) finally: runtime.close() ``` ```bash CAVEMAN_MODE=compress caveman start CAVEMAN_ENDPOINT=http://127.0.0.1:8787 DEMO_MODE=compress python quickstart.py ``` ```console capabilities {'mode': 'compress', 'persistent': True, 'recovery': True, 'retention_seconds': 86400} {"schema_version": 1, "status": "applied", "reason": "eligible", "transform_ids": ["caveman.engine.log.v1"], "replacement_count": 1, "reused_count": 0, "adapter": "langchain", …} {"schema_version": 1, "status": "reused", "reason": "eligible", "transform_ids": ["caveman.engine.log.v1"], "replacement_count": 1, "reused_count": 1, "adapter": "langchain", …} exact recovery SHA-256 10ca78813d5f173b897be30f06f6dd82367a1c11cdb221e1bf046e3d90c69f9d {'text': 'Deterministic fixture completed.', 'elapsed_ms': 40, 'usage': 'fixture; not provider usage', 'unchanged_history': True} ``` The `logical_call_id` and `attempt_id` fields vary per run and are cut here. `unchanged_history: True` is the assertion that the application's own message list was never rewritten, and the SHA-256 line is the recovered original hashed after paging it back in full. `DEMO_MODE=record` prints one `recorded` report, `DEMO_MODE=off` prints one `disabled` report. ## The other adapters Every file below is downloadable under `/examples/middleware/python/`. [Supported frameworks](/docs/sdk/middleware/frameworks) carries the accepted version range for each. ### OpenAI `with_caveman_openai` installs a Caveman transport under the client's httpx stack, so Chat Completions and Responses both route through it with the call site unchanged. `with_caveman_openai_tools` adds a tool loop that executes `caveman_retrieve` for you. ```python 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) ``` ### Anthropic The same shape for an `anthropic.Anthropic` client. The wrapped client keeps the `messages.create` signature. ```python 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) ``` ### Google Gen AI Google's client takes its transport at construction, so the adapter supplies one for the sync path and one for the async path, then wraps the client for tool registration. ```python 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() ``` ### LiteLLM `CavemanLiteLLM` is a client rather than a wrapper, because LiteLLM's entry point is a module function. `scope` moves to the call. ```python 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() ``` ### Strands `with_caveman_agent` rewrites the `Agent` keyword arguments. `with_caveman_model` wraps the model alone. ```python 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)) ``` ### Agno Same call, for `agno.agent.Agent`. `scope_from_run(run, namespace=...)` derives the scope from an Agno run object. ```python 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)) ``` ### CrewAI The options dict is the agent's own, so role, goal and backstory pass through untouched. `with_caveman_llm` wraps a bare LLM. ```python 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)) ``` ### AutoGen `with_caveman_agent` rewrites the `AssistantAgent` arguments, including its tool list. `component_runtimes` maps runtimes by key for a multi-agent team. ```python 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)) ``` ### Pydantic AI The adapter is a capability rather than a wrapper, so it attaches to the agent's capability list. ```python 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)]) ``` ### LlamaIndex `CavemanFunctionAgent` replaces `FunctionAgent`. `with_caveman_model` and `with_caveman_tools` cover graphs that build their own loop. ```python 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.") ``` ### MCP An MCP host owns the tool loop, so the adapter gives you the two halves separately: bind the source tool, then project its result into the view the model sees. The original stays in your hands. ```python 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 ``` ### ASGI For a FastAPI or Starlette service that fronts a provider API. The middleware compresses named routes. `resolve_context` returning `None` leaves a request alone, so the scope can only come from server state your authentication layer set. ```python 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) ``` ## Undo Pass the original options. Every `with_caveman_*` returns a new object and mutates nothing, so removing the call restores the previous behaviour. `mode="off"` on the runtime does the same without touching the call site.