Pydantic AI

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

import asyncio
import os
from pathlib import Path
from typing import List, Optional

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
    created_at: Optional[str] = None


class SearchHandoff(BaseModel):
    route_used: str
    has_more: bool = False
    next_cursor: Optional[str] = None
    tweets: List[TweetRow]
    summary: List[str]


async def main():
    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=SearchHandoff,
        system_prompt=(
            "Use Twexapi MCP. Call explore first, then twexapi_request with "
            "a catalog path. Preserve IDs and cursors."
        ),
    )

    result = await agent.run(
        "Search recent tweets about AI agents and return a structured handoff."
    )

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


asyncio.run(main())

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

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