LangChain

TwexAPI MCP를 통해 트윗 검색, 프로필, 팔로어 및 검토된 X 작업을 위한 LangChain 및 LangGraph Twitter API 에이전트를 구축합니다.

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의 엔드포인트 경로, 메서드 및 응답 필드를 반환합니다. 문서화된 query 및 body 필드만 twexapi_request에 전달하세요.

페이지네이션 경로는 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