Pydantic AI: a support agent with tools that dispatch to a different Manager

This example shows the Pydantic AI (edri.integrations.pydantic_ai) end to end: an agent's tools call other EDRI services as plain Events, the model calls two of them in parallel in the same turn, and the reply for one of them is resolved by a Worker on a different Manager entirely — none of that crossing needs any manual wiring.

• AgentManager hosts the agent and its SupportAgentWorker.

• OrderManager owns the actual order data and its own workers.

• A tool calls dispatch(OrderLookup(...)); the event is handled by OrderManager's worker, and the reply finds its way back to SupportAgentWorker's pipe automatically (ManagerBase tracks it, keyed on Event._id).

---

Events

from http import HTTPMethod
from typing import Optional

from edri.api.dataclass.api_event import api
from edri.dataclass.event import Event, event
from edri.dataclass.response import Response, response
from edri.integrations.pydantic_ai import ToolResponse


@response
class OrderStatusResponse(ToolResponse):
    status: str
    eta_minutes: Optional[int] = None

    @property
    def tool_result(self) -> str:
        if self.status == "not_found":
            return "No such order."
        if self.eta_minutes is not None:
            return f"Order is {self.status}, ETA {self.eta_minutes} minutes."
        return f"Order is {self.status}."


@event
class OrderLookup(Event):
    order_id: str
    response: OrderStatusResponse


@response
class ListAllOrdersResponse(ToolResponse):
    orders: str

    @property
    def tool_result(self) -> str:
        return self.orders


@event
class OrderListAll(Event):
    response: ListAllOrdersResponse


@response
class SupportAnswerResponse(Response):
    answer: str
    tool_used: bool = False


@api(url="/ask", method=HTTPMethod.POST)
class AskSupportAgent(Event):
    question: str
    response: SupportAnswerResponse

OrderStatusResponse/ListAllOrdersResponse subclass ToolResponse, not plain Response — that's what lets a tool call dispatch(...) and get back tool_result directly, while normal callers still see the typed fields.

---

OrderManager (owns the order data)

# order_manager.py
from edri.abstract import ManagerBase, request
from edri.abstract.worker import WorkerProcess
from events import OrderListAll, OrderLookup

# Stand in for a real backend (database, upstream API, ...).
ORDERS = {
    "A100": ("shipped", 45),
    "A200": ("processing", None),
    "A300": ("delivered", 0),
}


class OrderLookupWorker(WorkerProcess[OrderLookup]):
    def do(self) -> None:
        status, eta_minutes = ORDERS.get(self.event.order_id, ("not_found", None))
        self.event.response.status = status
        self.event.response.eta_minutes = eta_minutes
        self.event_send(self.event)


class OrderListAllWorker(WorkerProcess[OrderListAll]):
    def do(self) -> None:
        lines = [f"Order {oid}" for oid in sorted(ORDERS.keys())]
        self.event.response.orders = "\n".join(lines) if lines else "No orders found."
        self.event_send(self.event)


class OrderManager(ManagerBase):
    @request
    def lookup(self, event: OrderLookup) -> None:
        self.workers.start(event, OrderLookupWorker)

    @request
    def list_all(self, event: OrderListAll) -> None:
        self.workers.start(event, OrderListAllWorker)

OrderManager never even knows an agent is calling it — as far as it's concerned, OrderLookup/OrderListAll are just ordinary requests.

---

AgentManager (hosts the agent)

# agent_manager.py
import os
import threading
from logging import getLogger
from typing import Annotated, Any

from pydantic import Field
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider

from edri.abstract import ManagerBase, request
from edri.integrations.pydantic_ai import AgentBase, AgentDeps, BaseAgentWorker
from events import AskSupportAgent, OrderListAll, OrderLookup

# Order ids are always "A" followed by three digits (A100, A200, ...).
OrderId = Annotated[str, Field(pattern=r"^A\d{3}$")]

logger = getLogger(__name__)


class SupportAgent(AgentBase):
    output_mode = "text"

    def build_agent(self) -> Agent:
        self.tool_used = False
        model = OpenAIChatModel(
            self.model_name,
            provider=OpenAIProvider(
                base_url=os.environ["LLM_BASE_URL"],
                api_key=os.environ["LLM_API_KEY"],
            ),
        )
        agent = Agent(
            model,
            deps_type=AgentDeps,
            model_settings={"parallel_tool_calls": True},
            system_prompt=(
                "If the user gives a specific order id, call lookup_order_status "
                "with it. If they ask what orders exist or for the status of "
                "orders in general without an id, call list_all_orders first, "
                "then lookup_order_status for each id it returns."
            ),
        )

        # strict=False: see "Tool schemas and strict" in integrations/pydantic_ai.md — some
        # backends can hang on a strict schema for a call the model isn't
        # confident about yet (no id known). The `OrderId` pattern still helps
        # even without strict enforcement — see "Things to watch" below.
        @agent.tool(strict=False)
        def lookup_order_status(ctx: RunContext[AgentDeps], order_id: OrderId) -> Any:
            self.tool_used = True
            return ctx.deps.dispatch(OrderLookup(order_id=order_id))

        @agent.tool
        def list_all_orders(ctx: RunContext[AgentDeps]) -> Any:
            self.tool_used = True
            return ctx.deps.dispatch(OrderListAll())

        return agent

    def ask(self, question: str) -> tuple[str, bool]:
        self.tool_used = False
        result = self._run(question)
        return result.output, self.tool_used


class SupportAgentWorker(BaseAgentWorker[AskSupportAgent]):
    agents = [SupportAgent]
    agent_configs = {SupportAgent: {"model_name": "openai:gpt-4o"}}

    def do(self) -> None:
        logger.info(
            "SupportAgentWorker pid=%s thread=%s asking: %r",
            os.getpid(), threading.current_thread().name, self.event.question,
        )
        answer, tool_used = self.agent[SupportAgent].ask(self.event.question)
        self.event.response.answer = answer
        self.event.response.tool_used = tool_used
        self.event_send(self.event)


class AgentManager(ManagerBase):
    @request
    def ask(self, event: AskSupportAgent) -> None:
        self.workers.start(event, SupportAgentWorker)

---

Entry point

# main.py
from edri import EDRI

from agent_manager import AgentManager
from order_manager import OrderManager

if __name__ == "__main__":
    edri = EDRI()
    edri.add_component(OrderManager())
    edri.add_component(AgentManager())
    edri.start_api()
    edri.run()

---

Calling it

curl -X POST http://localhost:8080/ask -H 'Content-Type: application/json' \
  -d '{"question": "What is the status of order A100 and order A200?"}'

The model calls lookup_order_status twice — once per id — in the same turn; both dispatches run concurrently on SupportAgentWorker's pipe instead of queuing behind each other, and each is answered by a Worker on the other Manager:

{"data": {"answer": "Order A100 is shipped and has an ETA of 45 minutes. Order A200 is currently processing.", "toolUsed": true}}

A fully open-ended question instead ("What orders exist and what's their status?", no ids given) makes the model call list_all_orders first — there's a genuine data dependency, it needs real ids before it can look any of them up — and only then fan out parallel lookup_order_status calls once it has them:

{"data": {"answer": "Order A100 is shipped (ETA 45 minutes). Order A200 is processing. Order A300 is delivered.", "toolUsed": true}}

---

Things to watch

• A tool's target Event must reply with a ToolResponse. dispatch(...) raises a clear error otherwise, so this fails at the first real call, not silently.

• An Event with no declared response can never get a reply. dispatch(...) raises immediately rather than hanging out its 30s timeout.

• Per-call timeout. dispatch takes a keyword-only timeout (default 30s) — e.g. ctx.deps.dispatch(SlowLookup(...), timeout=5.0) for a tool hitting a known-slow downstream Worker.

• strict and guided decoding. pydantic-ai auto-promotes a simple tool schema to strict: true; some inference backends can hang indefinitely on it for a call the model isn't confident about yet (no valid argument value available). See Tool schemas and strict.

• Regex-constrained fields, e.g. OrderId. Since order ids all have the same shape, Annotated[str, Field(pattern=r"^A\d{3}$")] puts that pattern straight into the tool's JSON schema. Its value comes in two layers:

  • At generation — only if the backend enforces the schema while decoding (a guided-decoding backend, or OpenAI strict mode, which does accept a simple pattern like this). This example runs strict=False, so that layer is off here — the pattern is not constraining the model's tokens.

  • At parsing — always. When the model returns a tool call, pydantic-ai validates its arguments against the schema before invoking the tool, even with strict=False. A malformed order_id never reaches lookup_order_status; instead pydantic-ai sends the validation message ("String should match pattern '^A\d{3}$'") back to the model as a retry, and the model self-corrects. So the pattern earns its keep as a server-side guardrail with automatic model correction, not just a hint — even in the strict=False setup shown here.

• Several agents, several models. A worker isn't limited to one agent — agent_configs = {FastAgent: {"model_name": "..."}, DeepAgent: {"model_name": "..."}} gives each its own model; see How it fits together.