Pydantic AI

Verified 2026-08-19 — PASS, 4s. Session persistence: message_history round-trip.

Pydantic AI exposes the full typed message list, so persistence is one adapter call each way.

This is one of the two frameworks that appeared to fail in an earlier test run. It was never a framework fault — the transport underneath had a poisoned session. See sessions-and-swarm.md.

---

The runner

import sys, os, json, pathlib
sys.path.insert(0, os.path.expanduser("~/agenttests"))
import azcfg, hetool
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider
from pydantic_ai.messages import ModelMessagesTypeAdapter

SESS = pathlib.Path(os.path.expanduser("~/agenttests/pydanticai/sessions"))
SESS.mkdir(parents=True, exist_ok=True)

def get_working_folder() -> str:
    """Return the agent's current working folder (absolute path)."""
    return hetool.get_working_folder()

def run_shell(command: str) -> str:
    """Run a shell command in the working folder and return its output."""
    return hetool.run_shell(command)

def invoke(sid, message):
    c = azcfg.load()
    model = OpenAIChatModel(c["model"],
                            provider=OpenAIProvider(base_url=c["base_url"], api_key=c["api_key"]))
    agent = Agent(model, system_prompt="Be concise. Use tools for operational questions.",
                  tools=[get_working_folder, run_shell])

    f = SESS / f"{sid}.json"
    history = ModelMessagesTypeAdapter.validate_json(f.read_text()) if f.exists() else None
    result = agent.run_sync(message, message_history=history)
    f.write_bytes(ModelMessagesTypeAdapter.dump_json(result.all_messages()))
    return str(result.output).strip()

Note run_sync — no async needed, though agent.run is available if you prefer.

---

Engine definition

{
  "type": "process",
  "file": "/home/you/pydanticai/.venv/bin/python3",
  "nativeSession": true,
  "argsNew":    ["/home/you/he_adapter.py", "/home/you/pydanticai_session.py", "",            "{message}"],
  "argsResume": ["/home/you/he_adapter.py", "/home/you/pydanticai_session.py", "{sessionId}", "{message}"],
  "replyField": "result",
  "sessionIdField": "session_id",
  "approve": false,
  "workdir": "/home/you/work",
  "proxy": { "shape": "openai", "baseUrlEnv": "OPENAI_BASE_URL", "keyEnv": "OPENAI_API_KEY",
             "modelEnv": "OPENAI_MODEL", "model": "<route>|<provider model>" }
}

---

Adoption

1. Keep your agent, tools and result types as they are.
2. Persist result.all_messages() keyed on sid; pass it back as message_history.
3. Build OpenAIChatModel with an OpenAIProvider from azcfg.load().
4. Seal the engine, point external.engine at it, restart.

---

Gotchas

ModelMessagesTypeAdapter both ways — do not hand-roll JSON. The message list is typed (tool calls, tool returns, parts), and dump_json / validate_json preserve it exactly. Serialising it yourself loses structure and breaks resume in ways that surface much later.

dump_json returns bytes, hence write_bytes paired with read_text for the parse.

Provider goes through OpenAIProvider, not constructor kwargs on the model.

result.output, not result.data. The attribute was renamed; older examples on the web still use data and will raise.

Typed outputs still work. If your agent declares a Pydantic result type, result.output is that type — stringify it however your product needs before returning, since the adapter contract expects a string.