LangChain

Tweet araması, profiller, takipçiler ve incelenen X eylemleri için TwexAPI MCP aracılığıyla LangChain ve LangGraph Twitter API aracıları oluşturun.

TwexAPI’nin MCP sunucusu aracılığıyla bir LangChain Twitter API aracısı oluşturun. Tweetleri arayın, profilleri inceleyin, takipçi listelerini sayfalara ayırın ve yazma işlemlerini inceleyin. Tweet kimliklerini, zaman damgalarını, imleçleri ve yönlendirme hatalarını yazılan değerler olarak koruyun.

LangChain’i neden TwexAPI ile kullanmalısınız?

LangChain, TwexAPI araçlarını modellere, alıcılara, veritabanlarına ve uygulama hizmetlerine bağlar. LangGraph dayanıklı durum, devam ettirilebilir işler ve insan onayı ekler.

Kısa araç çağırma konuşmaları için LangChain’i kullanın. Arızalar, onaylar veya süreç yeniden başlatıldıktan sonra işin devam etmesi gerektiğinde LangGraph’ı kullanın. Her ikisi de aynı MCP araçlarını ve normalleştirilmiş devir sözleşmesini kullanır.

Önkoşullar

• Python 3.10 veya üzeri
• Bir TwexAPI API anahtarı
• Araç ve yapılandırılmış çıktı desteğine sahip, LangChain destekli bir model Public docs focus on API-key reads.

Genel X okumalarının X Developer kimlik bilgilerine ihtiyacı yoktur. TwexAPI ile kimlik doğrulaması yapın.

Kurulum

Tekrarlanabilir yapılar için uyumlu küçük aralıklar yükleyin.

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

Sırları kaynak kontrolü dışında saklayın.

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

TwexAPI MCP’yi bağlayın

LangChain, MCP istemcisini çalıştırır. TwexAPI, MCP sunucusunu “https://api.twexapi.io/mcp” adresinde çalıştırır. Sunucu, keşif için “explore”ı ve kimliği doğrulanmış çağrılar için twexapi_requesti kullanıma sunar.

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’ uzak MCP araçlarını yükler. Her imleci, rotayı ve yazma durumunu harici olarak kaydedin. İstemci varsayılan olarak vatansızdır.

MCP yanıt sözleşmesini koruyun

MCP, explore ile uç nokta yollarını, yöntemleri ve yanıt alanlarını döndürür. Yalnızca belgelenen query ve body alanlarını twexapi_request’e iletin.

Sayfalandırılmış rotalar next_cursor, has_next_page veya hasMore gibi imleç alanlarını döndürür. Her sayfada aynı sorgu ve filtreleri yeniden kullanın. Her imleci opak kabul edin.

Bir koşul doğru olduğunda sayfalandırmayı durdurun:

• Temsilci talep edilen toplamı toplar.
• has_more veya has_next_page yanlış olur.
• next_cursor eksik veya tekrarlanıyor.
• Yapılandırılmış sayfa sınırına ulaşıldı.

Sabit tweet_id veya user_id değerlerine göre tweet’leri ve kullanıcıları tekilleştirin.

Devam ettirilebilir bir temsilci aktarımını sürdürün

Konuşma geçmişi güvenilir bir iş veritabanı değildir. Yeniden denemeler, sayfalandırma ve aşağı akış araçları için gereken kalıcı değerler.

Kontrol listesinin tamamı için Agent MCP Handoff konusuna bakın.

Hata işlemeyi oluşturun

Durum kodlarını işle birlikte saklayın. Açık onay olmadan yazma eylemlerini asla yeniden denemeyin.

X eylemlerine insan onayı ekleyin

Salt okunur aracılar tweet’leri otomatik olarak arayabilir. Yazma özellikli temsilcilerin, göndermeden veya yanıtlamadan önce bir inceleme sınırına ihtiyacı vardır.

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(),
)

Üretimde kalıcı bir LangGraph denetim noktası kullanın. Beklenmeyen bir rota, hesap, hedef, metin veya medya içeren her türlü eylemi reddedin.

Dayanıklı LangGraph iş akışları oluşturun

Ayrı keşif, inceleme, yürütme ve depolama. Her harici aramadan sonra tamamlanan son düğümü, rotayı, yanıt kimliklerini, imleci ve yeniden deneme sayısını sürdürün.

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()

Birden fazla MCP sunucusunu bağlayın

Aracınız birden fazla MCP sağlayıcısına bağlandığında sunucu adlarına önek ekleyin.

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 temsilcisine yalnızca mevcut işi için gerekli araçları verin.

Paket versiyonları

Sonraki adımlar

• MCP Araçları
• Ajan MCP Aktarımı
• Python SDK’sı
• Gelişmiş Twitter Araması