---
title: "Pydantic AI"
description: "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 ルートを提供します。

| Boundary | Pydantic AI control | Twitter agent benefit |
| -------------------- | ---------------------- | -------------------------------------------- |
| MCP 接続 | `MCPServerStreamableHTTP` | 認証情報をプロセス内に保持 |
| 最終レスポンス | Pydantic `output_type` | 不正なツイート行とカーソルを拒否 |
| 書き込みアクション | 人間レビューステップ | 投稿、返信、フォロー前に一時停止 |
| 接続ライフサイクル | `async with agent` | 関連呼び出し間で1つの MCP セッションを再利用 |

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

## 前提条件

* 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"
```

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

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

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

エージェントは `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` を返してください。

## 型付きハンドオフを構築

<CardGroup cols={2}>
  <Card title="ツイート高度検索" icon="search">
    クエリ、ルート、ツイート 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 キーをエージェント出力外に保持してください。[Agent MCP Handoff](/mcp/agent-handoff) を参照してください。

## MCP 接続を再利用

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

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

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

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

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

## エラー処理

| Status | Pydantic AI decision |
| ------ | ---------------------------------------------- |
| `400` | リトライ前にリクエストを修正 |
| `401` | 停止し、認証情報を差し替え |
| `403` | アクセスまたはクレジット問題を報告 |
| `429` | バックオフし、カーソルを保持 |
| `5xx` | 安全な読み取りに上限付きバックオフでリトライ |

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

## Pydantic AI MCP か REST か

| 要件 | 選択肢 | 理由 |
| ------------------------------------------------ | --------------- | --------------------------------------------------- |
| モデルがツイートまたはプロフィール操作を選択 | Pydantic AI MCP | エージェントが `explore` でルートを発見 |
| アプリケーションコードが1つの既知ルートを呼び出し | [Python SDK](/sdks/python) | リクエストが決定論的 |
| 人間が X アクションをレビュー必須 | Pydantic AI MCP + 手動 REST | 実行前に一時停止 |
| モデルなしスケジュールエクスポート | [Prefect](/guides/prefect) または REST | モデル判断不要 |

## パッケージバージョン

| Package | Supported range |
| ------------------ | --------------- |
| 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)