---
title: "Pydantic AI"
description: "通过 TwexAPI MCP 构建 Pydantic AI Twitter API AI Agent，实现类型化推文搜索、资料查询、粉丝列表与经审核的 X 写操作。"
---

通过 TwexAPI 的 MCP 服务器构建 Pydantic AI Twitter API AI Agent。搜索推文、查看资料，并用 Pydantic 模型校验每条 durable handoff。审核每条拟议发帖、回复、点赞、关注或私信。

## 为何将 Pydantic AI 与 TwexAPI 搭配使用？

Pydantic AI 将模型工具调用与类型化 Python 输出结合。TwexAPI 通过 MCP 工具 `explore` 与 `twexapi_request` 提供 Twitter API 路由。

| 边界 | Pydantic AI 控制 | Twitter AI Agent收益 |
| -------------------- | ---------------------- | -------------------------------------------- |
| MCP 连接 | `MCPServerStreamableHTTP` | 凭证保留在进程内 |
| 最终响应 | Pydantic `output_type` | 拒绝畸形推文行与游标 |
| 写操作 | 人工审核步骤 | 发帖、回复或关注前暂停 |
| 连接生命周期 | `async with agent` | 相关调用复用同一 MCP 会话 |

推文搜索或资料 enrichment 各用一个类型化AI Agent，保持专注。

## 前置条件

* Python 3.10 或更高版本
* [TwexAPI API 密钥](https://twexapi.io/dashboard)
* 支持工具调用的 Pydantic AI 兼容模型
* 写操作所需的 Twitter cookie 或 `auth_token`

## 安装

```bash
python -m pip install "pydantic-ai[mcp]" python-dotenv
```

将密钥存放在源码控制之外。

```bash
export TWEXAPI_API_KEY="YOUR_API_KEY"
export ANTHROPIC_API_KEY="YOUR_ANTHROPIC_KEY"
```

## 构建类型化推文搜索AI Agent

创建AI Agent前先定义最终 handoff。Pydantic AI 按该 schema 校验模型输出。

```python
import asyncio
import os
from pathlib import Path
from typing import Literal

from dotenv import load_dotenv
from pydantic import BaseModel
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP


class TweetRow(BaseModel):
    tweet_id: str
    text: str
    author_username: str | None = None
    created_at: str | None = None
    url: str | None = None


class TweetSearchHandoff(BaseModel):
    query: str
    route_used: str
    tweets: list[TweetRow]
    has_more: bool
    next_cursor: str | None = None
    stop_reason: Literal["complete", "requested_limit", "cursor_stalled"]


async def main() -> None:
    load_dotenv()

    server = MCPServerStreamableHTTP(
        "https://api.twexapi.io/mcp",
        headers={"x-api-key": os.environ["TWEXAPI_API_KEY"]},
    )

    agent = Agent(
        "anthropic:claude-sonnet-4-20250514",
        toolsets=[server],
        output_type=TweetSearchHandoff,
        instructions=(
            "Use TwexAPI MCP for Twitter API requests. Call explore before twexapi_request. "
            "Preserve exact IDs and cursors. Never invent missing tweet fields. "
            "Ask for confirmation before read_only: false actions."
        ),
    )

    result = await agent.run(
        "Search 25 recent tweets about Pydantic AI MCP. "
        "Return the query, route, tweet rows, cursor state, "
        "and an explicit stop reason."
    )

    Path("twexapi-pydantic-ai-handoff.json").write_text(
        result.output.model_dump_json(indent=2),
        encoding="utf-8",
    )


asyncio.run(main())
```

AI Agent可发现 `explore` 与 `twexapi_request`。不知路由或参数形状时先调用 `explore`。

## 存储前校验字段

输出模型中保持以下源字段不变：

* 推文行：`tweet_id`、`text`、`author_username`、`created_at`、`url`
* 资料行：`user_id`、`username`、`name`、`description`、粉丝数
* 分页状态：`has_more`、`next_cursor` 或语言特定游标名
* 写回执：路由名、返回 tweet ID、确认状态

切勿将大 ID 转为浮点。仅在输出模型中将源 ID 映射为 `tweet_id` 或 `user_id`。

`has_more` 仍为 true 时继续空页。无游标或服务器重复游标时停止。返回 `cursor_stalled` 及已收集行数。

## 构建类型化 handoff

<CardGroup cols={2}>
  <Card title="推文高级搜索" icon="search">
    存储查询、路由、tweet ID、作者、URL、`has_more`、`next_cursor` 与停止原因。
  </Card>

  <Card title="查询用户资料" icon="user-round">
    存储 `user_id`、`username`、`name`、`description` 与粉丝数。
  </Card>

  <Card title="粉丝列表分页" icon="users">
    存储源用户名、粉丝行与游标检查点。
  </Card>

  <Card title="写入操作 (发推/点赞/关注)" icon="send">
    存储路由、预览文本、cookie 要求与审批记录。
  </Card>
</CardGroup>

API 密钥勿写入AI Agent输出。见 [Agent MCP Handoff](/mcp/agent-handoff)。

## 复用 MCP 连接

多次运行应共享连接时，包装相关调用。

```python
async def collect_two_search_pages(agent: Agent) -> None:
    async with agent:
        first_page = await agent.run(
            "Search 25 tweets about Pydantic AI MCP. Preserve the next cursor."
        )
        cursor = first_page.output.next_cursor
        if not first_page.output.has_more or cursor is None:
            return

        second_page = await agent.run(
            f"Continue tweet search for {first_page.output.query!r}. "
            f"Use explore, then twexapi_request with cursor {cursor!r}."
        )
        _ = second_page
```

## X 操作前要求审批

只读AI Agent可自动搜索推文。具备写能力的AI Agent在每次 X 操作前需人工决策。

```python
class WritePlan(BaseModel):
    action: str
    endpoint: str
    preview_text: str
    requires_human_confirmation: bool = True
```

调用 `read_only: false` 路由前停止AI Agent。用 [CLI](/sdks/cli) `--dry-run` 预览载荷，审批后通过 REST 或 [Python SDK](/sdks/python) 执行。

## 错误处理

| 状态 | Pydantic AI 决策 |
| ------ | ---------------------------------------------- |
| `400` | 重试前修正请求 |
| `401` | 停止并更换凭证 |
| `403` | 报告访问或额度问题 |
| `429` | 退避并保留游标 |
| `5xx` | 对有界退避重试安全读取 |

超时后切勿在未检查返回状态的情况下重建 pending 写操作。

## 选择 Pydantic AI MCP 或 REST

| 需求 | 选择 | 原因 |
| ------------------------------------------------ | --------------- | --------------------------------------------------- |
| 模型选择推文或资料操作 | Pydantic AI MCP | AI Agent用 `explore` 发现路由 |
| 应用代码调用已知单一路由 | [Python SDK](/sdks/python) | 请求保持确定性 |
| 人工须审核 X 操作 | Pydantic AI MCP + 手动 REST | 执行前暂停 |
| 定时导出无需模型 | [Prefect](/guides/prefect) 或 REST | 无需模型决策 |

## 包版本

| 包 | 支持范围 |
| ------------------ | --------------- |
| Python | `>=3.10` |
| `pydantic-ai` | `>=0.8` |
| `pydantic` | `>=2.7` |

## 下一步

* [MCP Tools](/mcp/tools)
* [Agent MCP Handoff](/mcp/agent-handoff)
* [LangChain](/guides/langchain)
* [Python SDK](/sdks/python)