Pydantic AI

TwexAPI MCP 経由で Pydantic AI Twitter API エージェントを構築し、型付きツイート検索、プロフィール、フォロワー、レビュー済み X アクションを実現する。

TwexAPI の MCP サーバー経由で Pydantic AI Twitter API エージェントを構築します。ツイート検索、プロフィール確認、すべての永続ハンドオフを Pydantic モデルで検証します。投稿、返信、いいね、フォロー、DM の各提案をレビューしてください。

TwexAPI と Pydantic AI を組み合わせる理由

Pydantic AI はモデルツール呼び出しと型付き Python 出力を組み合わせます。TwexAPI は MCP ツール explore と twexapi_request 経由で Twitter API ルートを提供します。

ツイート検索またはプロフィールエンリッチメントには1つの型付きエージェントを使用してください。各エージェントは焦点を絞ってください。

前提条件

• 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"

型付きツイート検索エージェントを構築

エージェント作成前に最終ハンドオフを定義します。Pydantic AI はこのスキーマに対してモデル出力を検証します。

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())

エージェントは explore と twexapi_request を発見します。ルートまたはパラメータ形状が不明な場合は先に explore を呼び出してください。

保存前にフィールドを検証

出力モデルでこれらのソースフィールドを変更しないでください。

• ツイート行: tweet_id, text, author_username, created_at, url
• プロフィール行: user_id, username, name, description, フォロワー数
• ページ状態: has_more, next_cursor, または言語固有カーソル名
• 書き込みレシート: ルート名、返されたツイート ID、確認ステータス

大きな ID を浮動小数点数にキャストしないでください。ソース ID は出力モデルで tweet_id または user_id にのみマップしてください。

has_more が true のまま空ページがあっても継続してください。カーソルが存在しない、またはサーバーがカーソルを繰り返す場合に停止し、収集行数とともに cursor_stalled を返してください。

型付きハンドオフを構築

API キーをエージェント出力外に保持してください。Agent MCP Handoff を参照してください。

MCP 接続を再利用

複数実行が1つの接続を共有する場合は関連呼び出しをラップします。

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 アクション前に承認を必須に

読み取り専用エージェントは自動的にツイートを検索できます。書き込み可能エージェントはすべての X アクション前に人間の判断が必要です。

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

read_only: false ルート呼び出し前にエージェントを停止してください。CLI --dry-run でペイロードをプレビューし、承認後 Python SDK または REST 経由で実行してください。

エラー処理

タイムアウト後、返されたステータスを確認せずに保留中の書き込みを再作成しないでください。

Pydantic AI MCP か REST か

パッケージバージョン

次のステップ

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