Pydantic AI

通过 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 路由。

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

前置条件

• Python 3.10 或更高版本
• TwexAPI API 密钥
• 支持工具调用的 Pydantic AI 兼容模型
• 写操作所需的 Twitter cookie 或 auth_token

安装

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

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

export TWEXAPI_API_KEY="YOUR_API_KEY"
export ANTHROPIC_API_KEY="YOUR_ANTHROPIC_KEY"

构建类型化推文搜索AI Agent

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

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

API 密钥勿写入AI Agent输出。见 Agent MCP Handoff。

复用 MCP 连接

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

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 操作前需人工决策。

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

调用 read_only: false 路由前停止AI Agent。用 CLI --dry-run 预览载荷，审批后通过 REST 或 Python SDK 执行。

错误处理

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

选择 Pydantic AI MCP 或 REST

包版本

下一步

• MCP Tools
• Agent MCP Handoff
• LangChain
• Python SDK