Skip to content
Redline
Esc
↑↓navigate↵open⌘Jpreview
On this page

Pydantic AI

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; anything else is your own tool loop.

1. Install and scaffold

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

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

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:

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:

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.

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:

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.)

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.

Was this page helpful?