---
title: Pydantic AI
description: Connect a Pydantic AI agent to Redline — experiments, production monitoring and runtime protection. The complete document, written to be followed by a coding agent inside the repository.
---

You are connecting a repository whose agent is built on **Pydantic AI** to the
Redline platform: experiments AND production monitoring. Follow these steps
exactly and do not modify the repository's existing agent code — the whole
integration is ONE new file.

## 0. Confirm the shape

Grep for `pydantic_ai` and `Agent(`. If present, this is **Shape A** and this
is the page. Pydantic AI is PATCHED by the SDK: the experiment's attached tools
appear in the agent by themselves, and its spans are emitted for you. If the
grep finds `langgraph` / `langchain.agents` instead, use the
[LangChain page](/integrations/langchain); anything else is
[your own tool loop](/integrations/own-tool-loop).

## 1. Install and scaffold

At the repository root, inside its virtualenv if it has one:

```bash
pip install redlineai-sdk
redline init
```

This writes `agents.py` — the doorway file. It must be named `agents.py` (or
live in an `agents/` package), **never `redline.py`**, which shadows the SDK on
`sys.path`. Nothing else to install: the SDK activates
`Agent.instrument_all()` itself.

## 2. Find the agent's entry point

Locate the function that RUNS the agent. Search for `.run(`, `.run_sync(`,
`.run_stream(` on the `Agent` instance, and for the `deps=` it is called with.
Read the signature. **Do NOT modify it.**

## 3. Wire it into `agents.py`

```python title="agents.py"
from redline import agent
from my_app.bank_support import support_agent, SupportDependencies   # unchanged

@agent(
    id="bank-support",
    name="Bank Support",
    description="Answers a customer's banking question and returns structured advice with a risk score.",
)
def run(task, ctx):
    result = support_agent.run_sync(task.prompt, deps=_deps())
    ctx.log(f"block_card={result.output.block_card} risk={result.output.risk}")
    return result.output.support_advice
```

- A **sync** entry point (`run_sync`) keeps a plain `def`. An **async** one
  (`await agent.run(...)`) gets `async def run` and awaits it — the SDK runs it
  on a persistent event loop it owns. **NEVER wrap it in `asyncio.run()`**:
  that builds and destroys a loop per run, and any async resource the agent
  keeps between calls then tears down into a closed loop.
- `id` is a kebab-case slug unique to this agent.
- `task.prompt` is the instructions. RETURN the final answer as a string.
- If the entry point takes messages, pass
  `[{"role": "user", "content": task.prompt}]`.
- If it streams (`run_stream`), consume the stream and return the whole text.
- An agent with an `output_type` does not *say* its answer — it calls the
  reserved `final_result` tool with it. Return the field you want graded (as
  above); the SDK also reads the structured result off the span for the
  transcript.
- Every file in `agents/` is imported by the CLI: export agents, start no servers.

## 4. Attached tools — do NOTHING

An experiment can attach MCP servers and, on every run, the project's Linux
machine as a `machine_run` shell tool. `redline dev` patches
`pydantic_ai.Agent.__init__` **before** it imports your files, so those tools
are merged in as your agent is constructed.

**Do not call `redline_tools`** — that would connect everything a second time.
Verify by counting: if the code declares 12 tools and the agent holds 13, the
injection landed. Attached SKILLS need nothing — they arrive inside `task.prompt`.

## 5. Three failures that are nearly guaranteed if you skip them

**5a. Tell the agent its tools exist.** A system prompt that describes a narrow
job produces "I have no way to do that" while `machine_run` sits unused. Because
the constructor is patched, the tools are in the list — add one line to the
`system_prompt` saying attached tools may be used when the task calls for them.

**5b. A tool that raises must not kill the run.** Pydantic AI retries a tool
that raises `ModelRetry`; any other exception propagates. Wrap your own tools
so an exception becomes the return text the agent reads and recovers from.

**5c. Do not drop the reasoning.** Pydantic AI's spans already carry model
calls and tool calls. `ctx.log(text)` is for what the spans do not say —
a structured decision, a count — as in §3. A run whose transcript shows only an
answer is a run nobody can judge.

## 6. Connect and verify

The agent's own environment (model keys) must load the way the repo normally
loads it. Then, in `.env` at the repo root or exported:

```bash
REDLINE_URL=https://tryredlineai.co
REDLINE_API_KEY=rl_…          # Agents page → Runner key
redline dev
```

`redline dev` reads that `.env` itself — **every variable in it**, not only
`REDLINE_*`, so model keys can live there too.

`redline dev` **must be running on this machine at all times** for anything to
work. It is the long-lived worker that connects this repository's agent to
Redline: experiments execute inside it, the agent shows ONLINE on the Agents
page only while it runs, asset snapshots for Protect upload through it,
policies and honeypots arrive and re-arm through it every 20 seconds, and dev
sessions stream to Monitor through it. If it stops, the agent goes offline.

What the two variables are:

- `REDLINE_URL` is the web address of the Redline platform. It is NOT an
  address of anything in this repository.
- `REDLINE_API_KEY` is the runner key minted on the Agents page. It is the only
  credential needed. **Do not commit it.**

On a server or VM, after you log out:

```bash
nohup redline dev > ~/redline-dev.log 2>&1 &
disown
```

It opens NO inbound port. Success is the line
`connected — bank-support registered and online`. Leave it running. It runs
several runs in parallel (`REDLINE_CONCURRENCY`, default 4).

## 7. When something is wrong

- **"no agents found"** → `agents.py` exports no `@agent`-decorated function.
- **Registration rejected** → the id collides; pick another slug.
- **The agent says it has no tools** → §5a, or the agent was constructed at
  import time in a module `redline dev` could not patch first. Construct it
  inside a function, or ensure `agents.py` imports it rather than the reverse.
- **Worker crashes with a closed event loop on the second run** → `asyncio.run()`
  in the doorway. Use `async def run` (§3).
- **Session shows only User/Agent rows, no model calls** → the SDK's
  `instrument_all()` did not run; confirm `redlineai-sdk` is the version
  `redline dev` is executing (`redline --version` inside the venv).
- **`ask_user` (if the agent has one) ENDS the run.** Return the questions as
  the answer rather than inventing replies.

Everything else: [known issues](/integrations/known-issues).

## 8. Monitoring — the same agent, watched in production

Nothing extra to write for dev: while `redline dev` is up, monitoring is on by
default (`REDLINE_MONITOR=1`) and every conversation the agent works ALSO
streams to **Monitor → Sessions**.

**8a. Instrumentation — nothing to install.** The SDK activates
`Agent.instrument_all()` when `redline dev` starts, so model calls, tokens and
tool calls are on every session. VERIFY after the first run: the session must
show model calls and a token count.

**For the DEPLOYED agent** — real users, no `redline dev` — decorate the
message handler with `@observe`. ONE decorator, no other monitoring code:

```python title="agents.py"
from redline import observe

@observe(agent="bank-support", session_arg="conversation_id",
         user_arg="user_id", input_arg="text")
async def handle_message(conversation_id: str, user_id: str, text: str) -> str:
    return support_agent.run_sync(text, deps=_deps()).output.support_advice
```

Each call becomes a session: the decorator records the input and the returned
answer, and every span Pydantic AI emits during the call lands in the SAME
session. Calls sharing `session_arg` land in one transcript. Config is
environment only — `REDLINE_API_KEY`, `REDLINE_URL`, `REDLINE_AGENT`;
**without the key the decorator is inert**. Errors are reported and re-raised.

**8b. Your application has to CALL `handle_message`.** A decorator only records
calls to the function it decorates. If the repository already handles inbound
messages — a route, a queue worker, a websocket loop — that code must call
`handle_message` for each message. One call-site change, the only one this
integration asks for. Without it, production sessions never appear.

**8c. It must return, not yield.** `@observe` checks
`inspect.iscoroutinefunction`, which is `False` for an async generator. If the
entry point streams, the handler drains the stream and returns the text.

(`create_monitor` remains as the explicit low-level client underneath — see
[monitoring](/integrations/monitoring).)

What the platform does with a session: it appears in Monitor → Sessions with
the waterfall; rule detectors match every span as it lands and the monitor
re-examines the session against every rule within minutes (Monitor →
Violations); when it closes, the intent miner reads it (Monitor → Intents).

## 9. Runtime protection — what to expect once connected (nothing to write)

While `redline dev` is up, the SDK ENFORCES the project's policies inside the
process. Pydantic AI speaks to the model through the OpenAI client, which the
SDK patches, so the gate sees every call. Configured in the console (**Protect
→ the agent → Policies**), live within ~20 seconds, no restart.

- **Governance rules** — a condition over the tool name and arguments, and a
  verdict. A denied call is stripped from the model's response before the
  framework can execute it and replaced with
  `[redline] The call to <tool> was denied by policy: <reason>`.
  `require_approval` holds it. `warn` lets the call run and appends
  `[redline] Policy note on <tool>: <reason>` to its result, so the agent
  reads why it was flagged; `log` lets it run and records it silently.
- **The guard classifier** — scores every user message and tool result for a
  prompt injection; above the block threshold the content is withheld.
- **Honeypots** — approved bait tools injected alongside the agent's own; a
  call to one is a recorded hijack signal, answered blandly.

Streaming responses pass through unjudged. Everything fails OPEN. Every
enforcement is recorded with its evidence in the console.