Pydantic AI

TwexAPI MCP üzerinden tipli tweet arama, profiller, takipçiler ve incelenen X eylemleri için Pydantic AI Twitter API agent'ları oluşturun.

TwexAPI’nin MCP sunucusu aracılığıyla bir Pydantic AI Twitter API agent’ı oluşturun. Tweet arayın, profilleri inceleyin ve her kalıcı devri bir Pydantic modeliyle doğrulayın. Önerilen her gönderi, yanıt, beğeni, takip veya doğrudan mesajı inceleyin.

Pydantic AI’yi TwexAPI ile neden kullanmalısınız?

Pydantic AI, model araç çağrılarını tipli Python çıktısıyla birleştirir. TwexAPI, MCP araçları explore ve twexapi_request aracılığıyla Twitter API rotaları sağlar.

Tweet araması veya profil zenginleştirmesi için tek bir tipli agent kullanın. Her agent’ı odaklı tutun.

Önkoşullar

• Python 3.10 veya daha yeni
• TwexAPI API anahtarı
• Araç çağırma destekli Pydantic AI uyumlu bir model
• Yazma işlemleri için Twitter çerezi veya auth_token

Kurulum

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

Gizlileri kaynak kontrolü dışında saklayın.

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

Tipli tweet arama agent’ı oluşturun

Agent’ı oluşturmadan önce son devri tanımlayın. Pydantic AI model çıktısını bu şemaya göre doğrular.

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

Agent explore ve twexapi_request araçlarını keşfeder. Agent bir rota veya parametre şeklini bilmediğinde önce explore çağırın.

Depolamadan önce alanları doğrulayın

Bu kaynak alanlarını çıktı modelinizde değiştirmeden tutun:

• Tweet satırları: tweet_id, text, author_username, created_at, url
• Profil satırları: user_id, username, name, description, takipçi sayıları
• Sayfa durumu: has_more, next_cursor veya dile özgü imleç adları
• Yazma makbuzları: rota adı, dönen tweet kimliği, onay durumu

Büyük kimlikleri asla kayan noktalı sayıya dönüştürmeyin. Kaynak kimliklerini yalnızca çıktı modelinizde tweet_id veya user_id olarak eşleyin.

has_more true kaldığında boş bir sayfada devam edin. İmleç yoksa veya sunucu bir imleci tekrarladığında durun. Toplanan satır sayısıyla cursor_stalled döndürün.

Tipli devir oluşturun

API anahtarlarını agent çıktısının dışında tutun. Bkz. Agent MCP Handoff.

MCP bağlantısını yeniden kullanın

Birden fazla çalıştırma tek bağlantıyı paylaşmalıysa ilgili çağrıları sarın.

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 eylemlerinden önce onay gerektirin

Salt okunur agent’lar tweet’leri otomatik arayabilir. Yazma yetenekli agent’lar her X eyleminden önce insan kararı gerektirir.

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

read_only: false rotalarını çağırmadan önce agent’ı durdurun. Yükleri CLI --dry-run ile önizleyin, ardından onaydan sonra REST veya Python SDK ile yürütün.

Hata yönetimi

Zaman aşımından sonra dönen durumu kontrol etmeden bekleyen bir yazma işlemini yeniden oluşturmayın.

Pydantic AI MCP mi REST mi?

Paket sürümleri

Sonraki adımlar

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