mirror of
https://github.com/pydantic/pydantic-ai-harness.git
synced 2026-07-21 10:55:35 +00:00
Pydantic AI's docs restructure gives Capabilities a top-level section (/ai/capabilities/) with a page per built-in capability. Update all /ai/core-concepts/capabilities/ links accordingly, and make the core built-ins explicit in the 'What goes where?' breakdown — naming and linking web search, web fetch, image generation, compaction, tool search, on-demand loading, thinking, and MCP, plus the full built-in list on the Capabilities overview. Merge together with the pydantic-ai docs restructure release: the new /ai/capabilities/ URLs only exist once that release is synced to pydantic.dev.
338 lines
16 KiB
Markdown
338 lines
16 KiB
Markdown
---
|
|
title: Code Mode
|
|
description: Wrap an agent's tools into a single sandboxed run_code tool so the model orchestrates many calls in one Python program instead of many round-trips.
|
|
---
|
|
|
|
# Code Mode
|
|
|
|
`CodeMode` replaces individual tool calls with a single sandboxed Python execution environment. Instead of the model issuing one tool call per action, it writes a Python program that calls your tools as functions -- with loops, conditionals, variables, and `asyncio.gather` -- all inside a sandboxed [Monty](https://github.com/pydantic/monty) runtime.
|
|
|
|
[Source](https://github.com/pydantic/pydantic-ai-harness/tree/main/pydantic_ai_harness/code_mode/)
|
|
|
|
## The problem
|
|
|
|
Standard tool calling costs one model round-trip per tool call. An agent that needs to fetch 10 items and process each one makes 11+ model calls -- slow, expensive, and heavy on context. The conversation history grows with every intermediate result, and everything runs sequentially unless the model deliberately batches parallel tool calls.
|
|
|
|
## The solution
|
|
|
|
`CodeMode` wraps the agent's tools into a single `run_code` tool. The model writes one Python snippet that orchestrates many tool calls locally: fan them out with `asyncio.gather`, filter and transform results in plain Python, and return only what matters.
|
|
|
|
| Standard tool calling | Code mode |
|
|
|---|---|
|
|
| 1 model call per tool | 1 model call for N tools |
|
|
| Sequential by default | Parallel via `asyncio.gather` |
|
|
| No local computation | Filter, transform, aggregate in code |
|
|
| Large conversation history | Compact -- fewer messages |
|
|
|
|
## Installation
|
|
|
|
Code mode requires the Monty sandbox, available via the `codemode` extra (the `code-mode` extra is an equivalent alias):
|
|
|
|
```bash
|
|
uv add "pydantic-ai-harness[codemode]"
|
|
```
|
|
|
|
## Usage
|
|
|
|
Construct an `Agent` with `CodeMode()` in its `capabilities`, then register tools as usual. Every tool becomes callable from inside `run_code`:
|
|
|
|
```python
|
|
from pydantic_ai import Agent
|
|
from pydantic_ai_harness import CodeMode
|
|
|
|
agent = Agent('anthropic:claude-sonnet-4-6', capabilities=[CodeMode()])
|
|
|
|
|
|
@agent.tool_plain
|
|
def get_weather(city: str) -> dict:
|
|
"""Get current weather for a city."""
|
|
return {'city': city, 'temp_f': 72, 'condition': 'sunny'}
|
|
|
|
|
|
@agent.tool_plain
|
|
def convert_temp(fahrenheit: float) -> float:
|
|
"""Convert Fahrenheit to Celsius."""
|
|
return round((fahrenheit - 32) * 5 / 9, 1)
|
|
|
|
|
|
result = agent.run_sync("What's the weather in Paris and Tokyo, in Celsius?")
|
|
print(result.output)
|
|
```
|
|
|
|
Inside a single `run_code` call, the model writes code like the following (illustrative -- the exact code the model emits will vary):
|
|
|
|
```python
|
|
import asyncio
|
|
|
|
paris, tokyo = await asyncio.gather(
|
|
get_weather(city='Paris'),
|
|
get_weather(city='Tokyo'),
|
|
)
|
|
paris_c = await convert_temp(fahrenheit=paris['temp_f'])
|
|
tokyo_c = await convert_temp(fahrenheit=tokyo['temp_f'])
|
|
{'paris': paris_c, 'tokyo': tokyo_c}
|
|
```
|
|
|
|
Both weather lookups run in parallel, the conversions run locally, and the whole thing collapses into one model round-trip instead of five.
|
|
|
|
## Selective tool sandboxing
|
|
|
|
By default, `CodeMode(tools='all')` sandboxes every tool. The `tools` field is a Pydantic AI `ToolSelector`, so you can control precisely which tools go through the sandbox. Tools that match the selector become callables inside `run_code`; non-matching tools stay visible to the model as regular tool calls.
|
|
|
|
```python
|
|
from pydantic_ai_harness import CodeMode
|
|
|
|
# By name -- only these tools are available inside run_code
|
|
CodeMode(tools=['search', 'fetch'])
|
|
|
|
# By predicate -- (ctx, tool_def) -> bool | Awaitable[bool]
|
|
CodeMode(tools=lambda ctx, td: td.name != 'dangerous_tool')
|
|
|
|
# By metadata -- combine with SetToolMetadata or a toolset's .with_metadata()
|
|
CodeMode(tools={'code_mode': True})
|
|
```
|
|
|
|
### Metadata-based selection
|
|
|
|
Use metadata when the decision should travel with a tool or toolset, rather than with one `CodeMode` instance. This suits shared toolsets: the toolset author tags the tools that are safe and useful to call from generated code, and each agent opts into that tag with `CodeMode(tools={...})`.
|
|
|
|
`CodeMode(tools={'code_mode': True})` uses the standard Pydantic AI [`ToolSelector`](/ai/api/pydantic-ai/tools/) metadata form. A tool is sandboxed when its `ToolDefinition.metadata` contains all of the selector's key-value pairs. Extra metadata on the tool is fine, and nested dictionaries are matched by deep inclusion.
|
|
|
|
The common pattern is to tag an entire toolset with `.with_metadata(...)`:
|
|
|
|
```python
|
|
from pydantic_ai import Agent
|
|
from pydantic_ai.toolsets import FunctionToolset
|
|
from pydantic_ai_harness import CodeMode
|
|
|
|
|
|
def search(query: str) -> str:
|
|
"""Search the web."""
|
|
return f'results for {query}'
|
|
|
|
|
|
def fetch(url: str) -> str:
|
|
"""Fetch a URL."""
|
|
return f'contents of {url}'
|
|
|
|
|
|
search_tools = FunctionToolset(tools=[search, fetch]).with_metadata(code_mode=True)
|
|
|
|
agent = Agent(
|
|
'anthropic:claude-sonnet-4-6',
|
|
toolsets=[search_tools],
|
|
capabilities=[CodeMode(tools={'code_mode': True})],
|
|
)
|
|
```
|
|
|
|
Here `search` and `fetch` are removed from the model-facing tool list and become callable functions inside `run_code`. Tools without `metadata['code_mode'] == True` stay visible as regular tool calls.
|
|
|
|
## Tool Search interaction
|
|
|
|
When you mark tools or whole toolsets `defer_loading=True` ([Tool Search](/ai/tools-toolsets/tools-advanced/#tool-search)), `CodeMode` keeps them out of `run_code` while they're undiscovered -- they pass straight through, so Tool Search drives them as usual (sent on the wire with `defer_loading` on providers with native tool search; otherwise dropped until discovered, with a `search_tools` tool alongside `run_code`). Once the model discovers a tool it comes back with `defer_loading=False`, and from then on `CodeMode` folds it into `run_code` like any other tool, so it's callable from generated code.
|
|
|
|
That fold-in grows `run_code`'s description, which invalidates the prompt-cache prefix once at the moment of discovery (turns with no discovery stay cache-warm). Two ways to avoid the bust:
|
|
|
|
- Pass `dynamic_catalog=True` to keep `run_code`'s description static across discoveries. The catalog of sandboxed-tool signatures moves into the agent instructions (as a dynamic [`InstructionPart`](/ai/api/pydantic-ai/messages/#pydantic_ai.messages.InstructionPart)) and newly-discovered tools are announced via [`ctx.enqueue`](/ai/api/pydantic-ai/tools/#pydantic_ai.tools.RunContext.enqueue) instead of by rebuilding the description:
|
|
|
|
```python
|
|
from pydantic_ai_harness import CodeMode
|
|
|
|
CodeMode(dynamic_catalog=True)
|
|
```
|
|
|
|
This pays off when paired with Tool Search: the tool-definitions block stays byte-stable so the prefix cache survives discoveries, at the cost of a larger (but cache-friendly) system prompt. With a fixed toolset and no Tool Search, the default keeps the system prompt shorter and is the better choice.
|
|
|
|
- To instead keep a Tool Search corpus fully native -- never folded into `run_code`, but not callable from inside it -- exclude it with a `tools` selector; corpus members carry `with_native` set to the managing native tool:
|
|
|
|
```python
|
|
from pydantic_ai_harness import CodeMode
|
|
|
|
CodeMode(tools=lambda ctx, td: td.with_native is None)
|
|
```
|
|
|
|
## Return values
|
|
|
|
The last expression in the snippet is automatically captured as the return value -- the model does not need to `print()`. An assignment stores a value in the REPL but does not return it. A final expression that evaluates to `None` is also treated as no result. Without a non-`None` final expression or print output, `run_code` returns `{}`. Put the assigned name on the final line:
|
|
|
|
```python
|
|
result = await get_weather(city='Paris')
|
|
result
|
|
```
|
|
|
|
Reserve `print()` for supplementary logging: printed text is surfaced separately, wrapped alongside the last-expression result.
|
|
|
|
| Scenario | Return |
|
|
|---|---|
|
|
| Non-`None` final expression with no print output | Last expression value |
|
|
| Final assignment or `None` result with no print output | `{}` |
|
|
| Print output with no final expression or a `None` result | `{'output': '<printed text>'}` |
|
|
| Print output with a plain, non-`None` final expression | `{'output': '<printed text>', 'result': <last expression>}` |
|
|
| Multimodal final expression with no print output | Returned natively for model processing |
|
|
| Print output with a multimodal final expression | List with printed text followed by native multimodal content |
|
|
|
|
## REPL state
|
|
|
|
State persists between `run_code` calls within the same agent run -- variables, imports, and function definitions carry over. Pass `restart: true` in the tool call to reset state.
|
|
|
|
## Observability
|
|
|
|
Nested tool calls inside `run_code` produce their own spans when instrumented with [Logfire](https://pydantic.dev/logfire) or any OpenTelemetry backend -- the easiest way to understand what code mode actually did, since each `run_code` span fans out into the tool calls the model issued from inside the sandbox. See the [Pydantic AI Logfire docs](/ai/integrations/logfire/) for setup.
|
|
|
|
The `run_code` tool return also carries metadata with every nested call, keyed by call id:
|
|
|
|
```python
|
|
from pydantic_ai import Agent
|
|
from pydantic_ai.messages import ToolReturnPart
|
|
from pydantic_ai_harness import CodeMode
|
|
|
|
agent = Agent('anthropic:claude-sonnet-4-6', capabilities=[CodeMode()])
|
|
|
|
|
|
@agent.tool_plain
|
|
def get_weather(city: str) -> dict:
|
|
"""Get current weather for a city."""
|
|
return {'city': city, 'temp_f': 72}
|
|
|
|
|
|
result = agent.run_sync("What's the weather in Paris?")
|
|
|
|
for msg in result.all_messages():
|
|
for part in msg.parts:
|
|
if isinstance(part, ToolReturnPart) and part.tool_name == 'run_code':
|
|
metadata = part.metadata or {}
|
|
tool_calls = metadata['tool_calls'] # dict[str, ToolCallPart]
|
|
tool_returns = metadata['tool_returns'] # dict[str, ToolReturnPart]
|
|
```
|
|
|
|
## In practice
|
|
|
|
A representative run wires `CodeMode` up against an MCP server and a web search and asks it to find the most-discussed Hacker News story across three feeds, pull the comment thread and the submitter's profile, and search the web for follow-up coverage. `CodeMode` collapses that into two `run_code` calls: the first fetches all three feeds in parallel via `asyncio.gather`, dedupes by id, filters by score, and ranks by comment count -- in plain Python; the second batches the three follow-up calls (`hn_get_thread`, `hn_get_user`, `duckduckgo_search`) together.
|
|
|
|
[](https://logfire-us.pydantic.dev/public-trace/84bcf123-2106-49da-9f6f-5c26395339bb?spanId=7650806a0785b946)
|
|
|
|
**[See the full Logfire trace ->](https://logfire-us.pydantic.dev/public-trace/84bcf123-2106-49da-9f6f-5c26395339bb?spanId=7650806a0785b946)** Each `run_code` span fans out into the tool calls the model issued from inside the sandbox.
|
|
|
|
## Filesystem and OS access
|
|
|
|
Sandboxed code runs with no access to the host's files, environment, or clock. Two parameters grant it access -- reach for them only when the agent's task genuinely needs the host.
|
|
|
|
Both parameters are fixed when the capability is built, so construct `CodeMode` per request to scope host access to that request.
|
|
|
|
### `mount` -- share host directories
|
|
|
|
Reach for `mount` when the agent works with real files: analyzing a dataset you've dropped in a folder and writing a report back, editing a checkout, or processing a batch of documents. Sandboxed `pathlib` code reads and writes under the mounted path. (For environment variables or the clock, use `os_access` instead.)
|
|
|
|
```python
|
|
from pydantic_ai import Agent
|
|
from pydantic_monty import MountDir
|
|
from pydantic_ai_harness import CodeMode
|
|
|
|
# The agent can read /work/data.csv and write /work/summary.md back to the host:
|
|
agent = Agent(
|
|
'anthropic:claude-sonnet-4-6',
|
|
capabilities=[CodeMode(mount=MountDir('/work', '/tmp/agent-workspace', mode='read-write'))],
|
|
)
|
|
```
|
|
|
|
A `MountDir` defaults to copy-on-write `mode='overlay'`: the sandbox reads host files and sees its own writes, but those writes do **not** reach the host. Pass `mode='read-write'` to persist them, or `mode='read-only'` to forbid writes. `mount` also accepts a list of `MountDir` for multiple mount points.
|
|
|
|
### `os_access` -- answer the sandbox's OS calls yourself
|
|
|
|
Reach for `os_access` when the agent needs environment variables, the current date and time, or filesystem behavior you control. Hand it a ready-made OS implementation (`AbstractOS`), or a callback that decides each call -- so you can inject just the secrets it needs, pin "now" for reproducible runs, or route file access to your own store.
|
|
|
|
```python
|
|
from pydantic_ai import Agent
|
|
from pydantic_monty import OSAccess
|
|
from pydantic_ai_harness import CodeMode
|
|
|
|
# Give the agent a fixed set of environment values:
|
|
agent = Agent(
|
|
'anthropic:claude-sonnet-4-6',
|
|
capabilities=[CodeMode(os_access=OSAccess(environ={'API_BASE': 'https://api.example.com'}))],
|
|
)
|
|
```
|
|
|
|
A callback receives each OS call and decides its fate:
|
|
|
|
```python
|
|
from pydantic_ai import Agent
|
|
from pydantic_monty import NOT_HANDLED
|
|
from pydantic_ai_harness import CodeMode
|
|
|
|
allowed_env = {'API_KEY': 'sk-...'}
|
|
|
|
|
|
def my_os(fn, args, kwargs):
|
|
if fn == 'os.getenv':
|
|
# Answer the call: allow-listed keys resolve, every other key reads back
|
|
# as None -- absent, exactly like a real unset variable.
|
|
return allowed_env.get(args[0])
|
|
# Refuse everything else: NOT_HANDLED makes the call fail in the sandbox.
|
|
return NOT_HANDLED
|
|
|
|
|
|
agent = Agent('anthropic:claude-sonnet-4-6', capabilities=[CodeMode(os_access=my_os)])
|
|
```
|
|
|
|
Your callback's return value decides the call's fate, and the two outcomes are easy to confuse:
|
|
|
|
- **Return any value** -- including `None`, `''`, or `0` -- and that becomes the result the sandbox sees. `os.getenv` returning `None` looks exactly like a normal unset variable, so the agent's code keeps running. This is how you *hide* something: answer with an empty value.
|
|
- **Return `NOT_HANDLED`** and the call is treated as unsupported: it raises inside the sandbox and the model gets a retry. This *refuses* a capability outright -- use it to block, not to say "no value". Returning `NOT_HANDLED` for a key the agent reasonably expects will burn retries.
|
|
|
|
!!! warning "Both expose the real host to model-written code"
|
|
`mount` and `os_access` hand model-generated Python real access to your filesystem and environment. Grant only what the task needs, and prefer constructing `CodeMode` per request so the granted access is scoped to that request.
|
|
|
|
!!! note "Monty-specific types"
|
|
These hooks use Monty's `AbstractOS`/`MountDir` types from `pydantic_monty`.
|
|
|
|
## Sandbox restrictions
|
|
|
|
Code runs inside [Monty](https://github.com/pydantic/monty), a sandboxed Python subset. Key restrictions:
|
|
|
|
- No class definitions.
|
|
- No third-party imports. Allowed stdlib modules: `sys`, `typing`, `asyncio`, `math`, `json`, `re`, `datetime`, `os`, `pathlib` (each must be imported before use).
|
|
- No wall-clock or timing primitives by default: `asyncio.sleep`, `datetime.datetime.now()`, `datetime.date.today()`, and the `time` module. `datetime.datetime.now()` / `datetime.date.today()` become available with an `os_access` handler (above); `asyncio.sleep` and `time` never do.
|
|
- No `import *`.
|
|
- Filesystem I/O needs an `os_access` handler or a `mount`; `os.getenv` / `os.environ` need an `os_access` handler.
|
|
- Tools requiring approval or with deferred (`CallDeferred`) execution are sandboxed like any other tool; without a `HandleDeferredToolCalls` (or equivalent) capability on the agent to resolve them inline, calling one from `run_code` raises an error that surfaces to the model as a retry.
|
|
|
|
## Agent spec (YAML/JSON)
|
|
|
|
`CodeMode` works with Pydantic AI's [agent spec](/ai/core-concepts/agent-spec/) feature for defining agents in YAML or JSON:
|
|
|
|
```yaml
|
|
# agent.yaml
|
|
model: anthropic:claude-sonnet-4-6
|
|
capabilities:
|
|
- CodeMode: {}
|
|
```
|
|
|
|
```python
|
|
from pydantic_ai import Agent
|
|
from pydantic_ai_harness import CodeMode
|
|
|
|
agent = Agent.from_file('agent.yaml', custom_capability_types=[CodeMode])
|
|
result = agent.run_sync('...')
|
|
print(result.output)
|
|
```
|
|
|
|
Pass `custom_capability_types` so the spec loader knows how to instantiate `CodeMode`. Arguments can be passed in the YAML too:
|
|
|
|
```yaml
|
|
capabilities:
|
|
- CodeMode:
|
|
tools: ['search', 'fetch']
|
|
max_retries: 5
|
|
```
|
|
|
|
## Further reading
|
|
|
|
- [Tool use via code](https://www.anthropic.com/engineering/code-execution-with-mcp) (Anthropic)
|
|
- [Code mode in production](https://blog.cloudflare.com/code-mode/) (Cloudflare)
|
|
- [Pydantic AI capabilities](/ai/capabilities/overview/)
|
|
|
|
## API reference
|
|
|
|
::: pydantic_ai_harness.CodeMode
|