# Pydantic AI: a support agent with tools that dispatch to a different Manager This example shows the {doc}`/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 ```python 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) ```python # 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) ```python # 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 ```python # 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 ```bash 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: ```json {"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: ```json {"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 {ref}`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 {ref}`how-it-fits-together`.