Let LangChain post to X
with one Python tool

# Store your key as an environment variable
export OPENTWEET_API_KEY="ot_your_key"

# Sanity check the key with GET /me
curl https://opentweet.io/api/v1/me \
  -H "Authorization: Bearer $OPENTWEET_API_KEY"

import os
import requests
from langchain_core.tools import tool

BASE = "https://opentweet.io/api/v1"

@tool
def post_tweet(text: str) -> str:
    """Publish a tweet to X immediately. Pass the full tweet text."""
    res = requests.post(
        f"{BASE}/posts",
        headers={
            "Authorization": f"Bearer {os.environ['OPENTWEET_API_KEY']}",
            "Content-Type": "application/json",
        },
        json={"text": text, "publish_now": True},
    )
    if res.status_code == 401:
        return "Error: bad or missing ot_ key"
    if res.status_code == 403:
        return "Error: your plan or account cannot post"
    if res.status_code == 429:
        return "Error: rate limited, back off and retry"
    if not res.ok:
        return f"Error {res.status_code}: {res.json().get('error', '')}"
    return f"Posted. id: {res.json().get('id')}"

from langchain.chat_models import init_chat_model
from langgraph.prebuilt import create_react_agent
from tools import post_tweet

model = init_chat_model("anthropic:claude-sonnet-4-5")

agent = create_react_agent(model, tools=[post_tweet])

result = agent.invoke({
    "messages": [
        {"role": "user", "content": "Write a short tweet about shipping fast and post it."}
    ]
})
print(result["messages"][-1].content)

@tool
def schedule_tweet(text: str, scheduled_date: str) -> str:
    """Schedule a tweet. scheduled_date is ISO 8601, e.g. 2026-07-10T09:00:00Z."""
    res = requests.post(
        f"{BASE}/posts",
        headers={
            "Authorization": f"Bearer {os.environ['OPENTWEET_API_KEY']}",
            "Content-Type": "application/json",
        },
        json={"text": text, "scheduled_date": scheduled_date},
    )
    return "scheduled" if res.ok else f"Error {res.status_code}"

@tool
def post_thread(tweets: list[str]) -> str:
    """Post a native X thread. Pass the tweets in order."""
    res = requests.post(
        f"{BASE}/posts",
        headers={
            "Authorization": f"Bearer {os.environ['OPENTWEET_API_KEY']}",
            "Content-Type": "application/json",
        },
        json={
            "is_thread": True,
            "thread_tweets": tweets,
            "thread_media": [[] for _ in tweets],
            "publish_now": True,
        },
    )
    return "posted thread" if res.ok else f"Error {res.status_code}"

import os
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain.chat_models import init_chat_model

client = MultiServerMCPClient({
    "opentweet": {
        "transport": "streamable_http",
        "url": "https://mcp.opentweet.io/mcp",
        "headers": {
            "Authorization": f"Bearer {os.environ['OPENTWEET_API_KEY']}",
        },
    }
})

tools = await client.get_tools()  # all 36 OpenTweet tools
agent = create_react_agent(init_chat_model("anthropic:claude-sonnet-4-5"), tools)

await agent.ainvoke({
    "messages": [{"role": "user", "content": "Schedule a tweet for tomorrow 9am UTC about our launch."}]
})