Pydantic AI

Pydantic AI 用 tool、依赖和 output type 定义有类型的 Agent。Dex adapter 把 Pydantic AI Agent 放进 durable session 中运行。它支持 Python。

何时使用

Agent 已经用 Pydantic AI 编写时，使用它。保留 Agent、它的 tool 和 capability，由 Dex 管理 session。这个 adapter 取代 Pydantic AI 自带的 durable execution 集成。

Session 如何运行

每个 RunModel Step 运行 Agent 时，会加上一个 capability，把每个 function tool 标记为 external tool。Pydantic AI 从不运行 external tool，因此模型请求 tool 之后，run 立即结束。

每个 tool call 在自己的 Step 中并行运行：read 或 idempotent write 用 CallTool，其他写操作用 CallSingleAttemptTool。需要审批的调用先在 AwaitToolApproval 中等待。Tool Step 通过 Agent 运行 tool 一次，因此 Agent 的 tool hook 在这里运行。下一个 RunModel 用记录下来的结果恢复 run。

查询订单的 turn 依次运行 RunModel、CallTool、ResumeTurn 和 RunModel。

安装

把 adapter 与它的 framework 一起安装。每个 adapter 都会带上 Dex Agents core。

$
pip install dex-agents-pydantic-ai "pydantic-ai-slim[openai]"

例子: agents/python/pydantic-ai/README.md

Example

这个 example 就是每个 framework 页面共用的订单客服 Agent。

Flow type 是 PydanticAIOrderSupport（Python）。完整 example 位于 examples/python/agent-harnesses。

Agent 与 tool

Tool 是 Agent 上普通的 Pydantic AI tool。Tool policy 指定每个 tool。Agent 推迟了模型检查，因此 Client 进程不需要 provider key。

pydantic_ai_order_support.pyPython
123456789101112131415161718192021222324252627
def build_pydantic_ai_order_support_flow(
    order_desk: OrderDesk, model: str | Model, slack_connection: slack.Connection | None
) -> AgentSessionFlow:
    """Builds the session Flow; ``model`` is a Pydantic AI model name, such as "openai:gpt-5.2"."""
    # The model is resolved at the first run, so a Client process needs no provider key.
    agent = Agent(model, instructions=ORDER_SUPPORT_INSTRUCTIONS, defer_model_check=True)

    # The tools are closures, so they call the OrderDesk passed in here.
    @agent.tool_plain
    def look_up_order(order_id: str) -> str:
        """Look up one order by its order ID and return its items and status."""
        return order_desk.describe_order(order_id)

    @agent.tool_plain
    def cancel_order(order_id: str, reason: str) -> str:
        """Cancel one order that has not shipped yet."""
        return order_desk.cancel_order(order_id, reason)

    return pydantic_ai_session_flow(
        agent,
        flow_type=PYDANTIC_AI_ORDER_SUPPORT_FLOW_TYPE,
        tool_sources=build_slack_tool_sources(slack_connection),
        tool_policies={
            "look_up_order": ToolPolicy(ToolEffect.READ),
            "cancel_order": ToolPolicy(ToolEffect.NON_IDEMPOTENT_WRITE),
        },
    )

例子: examples/python/agent-harnesses/agent_harness_examples/pydantic_ai_order_support.py

Worker

一个 Worker 为 example 中所有 harness 的 session Flow 提供服务。Agent session 的 Step 是 async 的，因此 Python Registry 允许 async handler。每个启动 session 或调用其 RPC 的进程，都要构建同一个 session Flow。

app.pyPython
1234567891011121314151617181920212223242526272829303132
        # Agent session Steps are async, so the Registry allows async handlers.
        self.registry = Registry(
            tuple(example.session_flow for example in self.examples), allow_async_handlers=True
        )
        config.blob_cache_dir.mkdir(parents=True, exist_ok=True)
        self.blob_cache = open_blob_cache(BlobCacheConfig(str(config.blob_cache_dir), 1 << 30))
        self.worker = AsyncWorker(
            self.registry,
            self.blob_cache,
            WorkerOptions(
                server_address=config.server_address,
                pull=PullWorkerOptions(max_concurrent_tasks=MAX_CONCURRENT_TASKS),
            ),
        )
        self.client = AsyncClient(
            self.registry, self.blob_cache, ClientOptions(server_address=config.server_address)
        )

# ...

def _new_pydantic_ai_example(
    config: AgentHarnessExamplesConfig, slack_connection: slack.Connection | None
) -> AgentHarnessExample:
    order_desk = OrderDesk()
    model = _pydantic_ai_model(config.pydantic_ai_model)
    flow = build_pydantic_ai_order_support_flow(order_desk, model, slack_connection)
    return AgentHarnessExample("pydantic-ai", flow, order_desk)

# ...

def _pydantic_ai_model(model: str) -> str | PydanticAIModel:
    return build_pydantic_ai_demo_model() if is_demo_model(model) else model

例子: examples/python/agent-harnesses/agent_harness_examples/app.py

Session route

每个 harness 都提供这同一组 route。这些 route 和 session token 的说明见 Session route。

http_routes.pyPython
1234567891011121314151617181920212223242526272829303132333435363738394041
    @blueprint.post("/sessions")
    async def start_session() -> Response:
        # The server picks the session ID, so a caller cannot claim an existing session.
        session_id = f"{harness_name}-order-support-{uuid.uuid4()}"
        await sessions.start_session(session_id)
        session_token = sign_session_id(app_state.config.session_key, session_id)
        return jsonify(session_id=session_id, session_token=session_token)

    @blueprint.post("/messages")
    async def send_user_message() -> Response:
        body, session_id = await session_request()
        accepted = await sessions.send_user_message(session_id, _required_string(body, "text"))
        return jsonify(accepted=accepted)

    @blueprint.post("/interrupts/resolve")
    async def resolve_interrupt() -> Response:
        body, session_id = await session_request()
        interrupt_id = _required_string(body, "interrupt_id")
        if not isinstance(approved := body.get("approved"), bool):
            raise BadRequest("approved must be a boolean")
        rejection_message = body.get("rejection_message")
        if rejection_message is not None and not isinstance(rejection_message, str):
            raise BadRequest("rejection_message must be a string")
        accepted = await sessions.resolve_interrupt(
            session_id, interrupt_id, approved=approved, rejection_message=rejection_message
        )
        return jsonify(accepted=accepted)

    @blueprint.get("/snapshot")
    async def get_agent_snapshot() -> Response:
        session_id = request.args.get("session_id", "")
        if not session_id:
            raise BadRequest("session_id is required")
        authorize_session(session_id)
        try:
            snapshot = await read_agent_event_stream_snapshot(
                app_state.client, example.session_flow, session_id
            )
        except FlowNotActiveOrNotFoundError as error:
            raise NotFound("the agent session is missing or no longer retained") from error
        return jsonify(snapshot.to_json())

例子: examples/python/agent-harnesses/agent_harness_examples/http_routes.py

在本地运行

启动 Dex 与 example

在第一个终端中，先渲染 session graph，再启动 Dex。Dex Web 启动时读取这个 graph，用它显示 session 和审批控件。

$$$$$
cd examples/python/agent-harnesses
uv sync --locked
mkdir -p definitions
uv run --frozen dex-agents-session-graph --flow-type PydanticAIOrderSupport \
  --permission order-support.operate --out definitions/pydantic-ai.json
dexcli dev --flow-rendering-dir ./definitions

例子: examples/python/agent-harnesses/README.md

dexcli dev 会一直在前台运行。在第二个终端中启动 example：

$$
cd examples/python/agent-harnesses
uv run --frozen python -m agent_harness_examples

例子: examples/python/agent-harnesses/README.md

应用监听 127.0.0.1:8081。Dex Web 位于 127.0.0.1:8802。

没有设置 DEX_PYDANTIC_AI_MODEL 时，Agent 使用 mock/order-support，这是一个不需要 key 的本地 demo model。收到 "look up order 1001" 时它调用 look_up_order，收到 "cancel order 1002" 时它调用 cancel_order。

变量	默认值	含义
DEX_PYDANTIC_AI_MODEL	mock/order-support	Pydantic AI 模型名称，例如 openai:gpt-5.2。
OPENAI_API_KEY	无	Provider key，使用 OpenAI 模型时需要。
DEX_AGENT_SESSION_KEY	随机 key	签发 session token 的 key。

调用 session

启动一个 session，让 Agent 取消订单 1002，然后批准这次调用：

$$$$$$$
base=http://127.0.0.1:8081/products/agent-harnesses/pydantic-ai
session=$(curl -s -X POST "$base/sessions" -H 'Content-Type: application/json' -d '{}')
id=$(echo "$session" | jq -r .session_id)
token=$(echo "$session" | jq -r .session_token)
curl -s -X POST "$base/messages" -H 'Content-Type: application/json' \
  -H "X-Agent-Session-Token: $token" -d "{\"session_id\": \"$id\", \"text\": \"Cancel order 1002\"}"
until interrupt=$(curl -s "$base/snapshot?session_id=$id" -H "X-Agent-Session-Token: $token" \
  | jq -re '.snapshot.pending_interrupts[0].interrupt_id'); do sleep 1; done
curl -s -X POST "$base/interrupts/resolve" -H 'Content-Type: application/json' \
  -H "X-Agent-Session-Token: $token" \
  -d "{\"session_id\": \"$id\", \"interrupt_id\": \"$interrupt\", \"approved\": true}"

例子: examples/python/agent-harnesses/README.md

循环会反复读取 snapshot，直到 Agent 请求运行 cancel_order。这次调用先等待审批，批准后只运行一次。再读一次 snapshot 即可看到 Agent 的回答。

在 Dex Web 中审批

也可以不执行最后一条命令，改在 Dex Web 中审批，步骤见在 Dex Web 中审批。

连接 Connector

Agent 可以把 Connector 的 operation 当作 tool 调用，见把 Connector 用作 tool。这个 example 加入了 Slack tool slack_post_thread_reply，用来在 Slack thread 中回复。

提供 tool

安装 dex-agents 的 connectors extra，以及 Connector 包，例如 dex-connector-slack。安装命令见 Connector tools。

Worker 从项目配置中打开 Slack connection support-slack。connector_tool 把 operation 变成 tool，由 ConnectorToolSource 持有。没有这个 connection 时，agent 不会得到 Slack tool。

slack_connector_tools.pyPython
1234567891011121314151617181920212223242526272829303132
def build_slack_tool_sources(connection: slack.Connection | None) -> list[ToolSource]:
    """Returns the Slack tool source, or no source when Slack is not configured."""
    if connection is None:
        return []
    return [
        ConnectorToolSource([
            # The manifest does not ask for approval; this agent asks before every post.
            connector_tool(
                connection.post_thread_reply(), connection.reference, requires_approval=True
            ),
        ]),
    ]

# ...

async def open_slack_project_connection(
    environment: Mapping[str, str] = os.environ,
) -> SlackProjectConnection | None:
    # ...
    # An incomplete or invalid configuration raises here, so the Worker fails at startup.
    try:
        project = await load_from_environment(environment)
    except FileNotFoundError:
        _log_slack_not_set_up("the connector configuration file does not exist")
        return None
    # ...
    try:
        connection = slack.new_project_connection(project, SLACK_CONNECTION_NAME)
    except BaseException:
        await project.close()
        raise
    return SlackProjectConnection(project, connection)

例子: examples/python/agent-harnesses/agent_harness_examples/slack_connector_tools.py

Worker 在启动时打开一次 connection，每个 agent 都使用它：

__main__.pyPython
123456789
async def main() -> None:
    config = AgentHarnessExamplesConfig.from_env()
    slack_setup = await open_slack_project_connection()
    try:
        app_state = AgentHarnessExamplesApp(config, slack_setup.connection if slack_setup else None)
        # ...
    finally:
        if slack_setup is not None:
            await slack_setup.aclose()

例子: examples/python/agent-harnesses/agent_harness_examples/__main__.py

Agent 把 tool source 和自己的 tool 一起传入：

pydantic_ai_order_support.pyPython
123456
    return pydantic_ai_session_flow(
        agent,
        flow_type=PYDANTIC_AI_ORDER_SUPPORT_FLOW_TYPE,
        tool_sources=build_slack_tool_sources(slack_connection),
        # ...
    )

例子: examples/python/agent-harnesses/agent_harness_examples/pydantic_ai_order_support.py

调用如何运行

每次调用 slack_post_thread_reply 都要先经人批准，然后在自己的 CallSingleAttemptTool Step 中只运行一次。原因见 Slack Connector tool。

Adapter 把 Connector tool 作为 ExternalToolset 传给每次 run。Pydantic AI 从不运行 external tool，所以 Connector tool call 会结束这次 run，调用在它的 tool Step 中运行。

配置 connection

Worker 从哪里找到 connection support-slack、Slack app 需要什么，见 Connection。

试用时，先设置 DEX_CONNECTOR_CONFIG_FILE 再启动 example。然后让 agent 在 Slack thread 中回复，并给出 channel ID 和 thread timestamp：

Reply in Slack thread C0123456789 1710000000.000100: Order 1002 ships on Thursday.

这次发帖和 cancel_order 一样，会先等待审批。

限制

Pydantic AI 自己发给模型的重试，会在同一个 RunModel Step 中再调用一次模型。例如 tool 参数无效，或 output 校验失败。
Output function 在 RunModel 内运行，至少运行一次。请把写操作放进 tool。
由 provider 运行的 tool，例如 WebSearchTool，会在 RunModel 重跑时重复执行。
Deferred tool 返回给模型的是文本。图片和文件会丢失。
只在运行时才请求审批的 tool，需要一个带 requires_approval=True 的 ToolPolicy。
UsageLimits 和 tool 重试预算按 Step 计算，而不是按 turn 计算。

Adapter README 列出了全部限制。Server 的限制参见 Limits for long sessions。

何时使用​

Session 如何运行​

安装​

Example​

Agent 与 tool​

Worker​

Session route​

在本地运行​

启动 Dex 与 example​

调用 session​

在 Dex Web 中审批​

连接 Connector​

提供 tool​

调用如何运行​

配置 connection​

限制​

相关内容​

何时使用

Session 如何运行

安装

Example

Agent 与 tool

Worker

Session route

在本地运行

启动 Dex 与 example

调用 session

在 Dex Web 中审批

连接 Connector

提供 tool

调用如何运行

配置 connection

限制

相关内容