Give a Pydantic AI agent a Twitter tool

import asyncio
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP

opentweet = MCPServerStreamableHTTP(
    "https://mcp.opentweet.io/mcp",
    headers={"Authorization": "Bearer ot_your_key"},
)

agent = Agent(
    "openai:gpt-4o",
    toolsets=[opentweet],
    instructions="You publish and schedule on X. Prefer scheduling unless told to post now.",
)

async def main():
    async with agent:
        result = await agent.run(
            "Write a 3-post thread about the 2.0 release and schedule it for 9am UTC tomorrow."
        )
        print(result.output)

asyncio.run(main())

import os
import httpx
from pydantic_ai import Agent

agent = Agent("openai:gpt-4o", instructions="You write and publish short posts on X.")

@agent.tool_plain
async def post_to_x(text: str, scheduled_date: str | None = None) -> str:
    """Publish a post to X, or schedule it. scheduled_date is ISO 8601, for example 2026-09-02T09:00:00Z."""
    body = {"text": text}
    if scheduled_date:
        body["scheduled_date"] = scheduled_date
    else:
        body["publish_now"] = True

    async with httpx.AsyncClient(timeout=30) as client:
        res = await client.post(
            "https://opentweet.io/api/v1/posts",
            headers={"Authorization": f"Bearer {os.environ['OPENTWEET_API_KEY']}"},
            json=body,
        )
    res.raise_for_status()
    return res.text

curl -X POST https://opentweet.io/api/v1/posts \
  -H "Authorization: Bearer ot_your_key" \
  -H "Content-Type: application/json" \
  -d '{"text":"The agent wrote this in Python and the scheduler sent it.","scheduled_date":"2026-09-02T09:00:00Z"}'