LangChain

TwexAPI MCP 経由で LangChain と LangGraph Twitter API エージェントを構築し、ツイート検索、プロフィール、フォロワー、レビュー済み X アクションを実現する。

TwexAPI の MCP サーバー経由で LangChain Twitter API エージェントを構築します。ツイート検索、プロフィール確認、フォロワーリストのページネーション、書き込みアクションのレビューが可能です。ツイート ID、タイムスタンプ、カーソル、ルートエラーを型付き値として保持してください。

TwexAPI と LangChain を組み合わせる理由

LangChain は TwexAPI ツールをモデル、リトリーバー、データベース、アプリケーションサービスに接続します。LangGraph は永続状態、再開可能なジョブ、人間の承認を追加します。

短いツール呼び出し会話には LangChain を使用します。障害、承認、プロセス再起動後に作業を再開する必要がある場合は LangGraph を使用します。両方とも同じ MCP ツールと正規化ハンドオフ契約を使用します。

前提条件

• Python 3.10 以降
• TwexAPI API キー
• ツール呼び出しと構造化出力をサポートする LangChain 対応モデル Public docs focus on API-key reads.

公開 X の読み取りには X Developer 認証情報は不要です。TwexAPI で認証してください。

インストール

再現可能なビルドのため、互換性のあるマイナーレンジをインストールします。

python -m pip install --upgrade \
  "langchain>=1.0" \
  "langchain-mcp-adapters>=0.2" \
  langchain-anthropic \
  langgraph \
  python-dotenv

シークレットはソース管理外に保存してください。

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

TwexAPI MCP に接続

LangChain が MCP クライアントを実行します。TwexAPI は https://api.twexapi.io/mcp で MCP サーバーを実行します。サーバーは発見用に explore、認証済み呼び出し用に twexapi_request を公開します。

import asyncio
import os
from pathlib import Path
from typing import Literal

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from pydantic import BaseModel


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",
        "page_cap",
    ]


async def main() -> None:
    load_dotenv()

    client = MultiServerMCPClient(
        {
            "twexapi": {
                "transport": "streamable_http",
                "url": "https://api.twexapi.io/mcp",
                "headers": {"x-api-key": os.environ["TWEXAPI_API_KEY"]},
            },
        }
    )
    tools = await client.get_tools()

    agent = create_agent(
        model="anthropic:claude-sonnet-4-20250514",
        tools=tools,
        response_format=TweetSearchHandoff,
        system_prompt=(
            "Use TwexAPI 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.ainvoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": (
                        "Search 25 recent tweets about LangChain MCP. "
                        "Return the query, route, tweet rows, cursor state, "
                        "and an explicit stop reason."
                    ),
                }
            ]
        }
    )
    handoff = result["structured_response"]
    Path("twexapi-langchain-handoff.json").write_text(
        handoff.model_dump_json(indent=2),
        encoding="utf-8",
    )


asyncio.run(main())

MultiServerMCPClient はリモート MCP ツールを読み込みます。すべてのカーソル、ルート、書き込みステータスを外部に保存してください。クライアントはデフォルトでステートレスです。

MCP レスポンス契約を保持

MCP は explore からエンドポイントパス、メソッド、レスポンスフィールドを返します。twexapi_request にはドキュメント化された query と body フィールドのみを渡してください。

ページネーションルートは next_cursor、has_next_page、hasMore などのカーソルフィールドを返します。各ページで同じクエリとフィルターを再利用し、各カーソルは不透明な値として扱います。

次のいずれかが真になったらページネーションを停止します。

• エージェントが要求総数を収集した。
• has_more または has_next_page が false になった。
• next_cursor が欠落または繰り返された。
• 設定されたページ上限に達した。

安定した tweet_id または user_id 値でツイートとユーザーを重複排除します。

再開可能なエージェントハンドオフを保持

会話履歴は信頼できるジョブデータベースではありません。リトライ、ページネーション、下流ツールに必要な値を永続化してください。

完全なチェックリストは Agent MCP Handoff を参照してください。

エラー処理を構築

ジョブと一緒にステータスコードを保存してください。明示的な承認なしで書き込みアクションをリトライしないでください。

X アクションに人間の承認を追加

読み取り専用エージェントは自動的にツイートを検索できます。書き込み有効エージェントは投稿または返信前にレビュー境界が必要です。

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model="anthropic:claude-sonnet-4-20250514",
    tools=tools,
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "twexapi_request": {
                    "allowed_decisions": ["approve", "reject"],
                }
            }
        )
    ],
    checkpointer=InMemorySaver(),
)

本番では永続 LangGraph チェックポインターを使用してください。予期しないルート、アカウント、ターゲット、本文、メディアのアクションは拒否してください。

永続 LangGraph ワークフローを構築

発見、レビュー、実行、保存を分離します。各外部呼び出し後に最後に完了したノード、ルート、レスポンス ID、カーソル、リトライ回数を永続化します。

from langgraph.graph import START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition

def call_model(state: MessagesState):
    return {"messages": model.bind_tools(tools).invoke(state["messages"])}

builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_node(ToolNode(tools))
builder.add_edge(START, "call_model")
builder.add_conditional_edges("call_model", tools_condition)
builder.add_edge("tools", "call_model")
graph = builder.compile()

複数 MCP サーバーに接続

エージェントが複数の MCP プロバイダーに接続する場合はサーバー名にプレフィックスを付けます。

client = MultiServerMCPClient(
    {
        "twexapi": {
            "transport": "streamable_http",
            "url": "https://api.twexapi.io/mcp",
            "headers": {"x-api-key": os.environ["TWEXAPI_API_KEY"]},
        },
        "docs": {
            "transport": "streamable_http",
            "url": "https://docs.twexapi.io/mcp",
        },
    },
    tool_name_prefix=True,
)

TwexAPI エージェントには現在のジョブに必要なツールのみを付与してください。

パッケージバージョン

次のステップ

• MCP Tools
• Agent MCP Handoff
• Python SDK
• Advanced Twitter Search