---
title: "How to add token compression to Pydantic AI agents"
description: "Your Pydantic AI agent keeps its model, tools, dependencies and output types. Caveman goes in as one more entry in capabilities=[...], plus a small local runtime. For the background, read [Caveman vs "
canonical: https://caveman.ac/switch/pydanticai
last-updated: 2026-10-04
---

# How to add token compression to Pydantic AI agents

Your Pydantic AI agent keeps its model, tools, dependencies and output types. Caveman goes in as one more entry in `capabilities=[...]`, plus a small local runtime. For the background, read [Caveman vs Pydantic AI](/compare/pydanticai).

## What you need

- Python 3.13 or newer and `pydantic-ai-slim >=2.42 <3`. Outside that range the capability declines, reports `unsupported_version`, and passes requests through untouched.
- A model built on Pydantic AI's `OpenAIChatModel` or `AnthropicModel`. Other model classes pass through untouched and report `unsupported_provider`.
- Tools that return plain strings. A dict or Pydantic model return is sent as is.
- Node.js 22.13 or newer for the `caveman` CLI that runs the runtime.
- An agent that answers in plain text and whose tools return a lot of text. The recovery tool is only offered on text-output runs with automatic tool choice, so start there rather than on a structured `output_type` agent.

## Step 1: Record a baseline

Run one representative task a few times as it is today, on a fixed model. Save the output, whether it passed your check, and `result.usage()` for each run.

## Step 2: Install the packages

```bash
pip install caveman-sdk "caveman-middleware[pydantic-ai]==0.1.0a1"
```

`caveman-middleware` only has prereleases so far, so pin the version or pass `--pre`. The runtime client lives in `caveman-sdk` as `caveman_cloud.middleware`.

## Step 3: Start the local runtime

```bash
npm install -g @caveman-ai/cli && caveman setup --install
CAVEMAN_MODE=compress caveman start
```

Check it from another shell:

```bash
curl -s http://127.0.0.1:8787/caveman/v1/middleware/capabilities | jq '{mode, persistent, recovery, retention_seconds}'
```

You want `"mode": "compress"` and `"recovery": true`. Plain `caveman start` runs `record`, which measures and changes nothing.

## Step 4: Add the capability

The docs example is one line on your `Agent`:

```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)])
```

Around it you need a runtime and a scope. The runtime is created once per process. `CavemanCapability` accepts the regular `MiddlewareRuntime` and converts it to an async view itself:

```python
from dataclasses import asdict
from caveman_cloud.middleware import MiddlewareRuntime, Scope

runtime = MiddlewareRuntime(endpoint="http://127.0.0.1:8787", mode="compress",
    deadline_ms=500, on_report=lambda report: print(asdict(report)))
if runtime.mode != "off":  # ready() raises in off mode
    runtime.ready()
```

The scope is four strings, `Scope(namespace, session_id, branch_id, cache_epoch)`, and it keys the stored originals. A recovery handle only resolves under the same four, so tie `session_id` to your conversation. You can pass a fixed `Scope`, or a function that receives the run context and is resolved once per run. The adapter ships `scope_from_run(ctx, namespace=...)`, which uses the run's native `conversation_id`. Pydantic AI generates one on the first run and carries it through `message_history`; pass `conversation_id=...` to `agent.run` to use your own:

```python
from caveman_middleware.pydantic_ai import CavemanCapability, scope_from_run

agent = Agent(model, tools=tools, capabilities=[CavemanCapability(
    runtime=runtime, scope=lambda ctx: scope_from_run(ctx, namespace="support"))])
```

Keep `namespace` to your tenant or app, never shared across end users. If the model decides to recover an original, that is one extra tool call, so leave headroom in any request limits you set.

## Step 5: Check it worked

The `on_report` callback prints one decision per model request:

- First request: `skipped` with `no_candidate`. No tool has returned yet.
- After the first large tool return: `applied`, with a transform id such as `caveman.engine.log.v1`.
- Later requests: `reused`. The same shortened bytes go out again, so the provider's cached prefix holds.

Then compare against the baseline. The agent's messages still hold the full tool returns. The task passes the same check. Provider input tokens in `result.usage()` are lower across the whole run, with any recovery calls counted in.

If nothing shrinks, read the reason. `runtime_unavailable` means the endpoint did not answer. `recorded` means the runtime is in record mode. `unsupported_version` means `pydantic-ai-slim` is outside the gate. `unsupported_provider` means the model is not `OpenAIChatModel` or `AnthropicModel`. Requests with structured output go out uncompressed.

## How to roll back

- Remove `CavemanCapability(...)` from the `capabilities` list. Nothing else changed.
- Or construct the runtime with `mode="off"`. The capability reports `disabled` and sends the original messages.

Call `runtime.close()` at shutdown. The runtime process keeps stored originals for 24 hours after their last use, so conversations already in flight can still recover.

## Next steps

Judge the change on finished tasks: [how to measure agent cost](/guides/measure-agent-cost). If you trace with Logfire, compare usage there per run. The [Python middleware docs](https://docs.caveman.so/docs/sdk/middleware/python) cover the runtime options, and the [deployment page](https://docs.caveman.so/docs/sdk/middleware/deployment) covers running the runtime beside a deployed app.