CrewAI

Tweet araması, profiller, takipçiler ve yazılan aktarımlar için TwexAPI MCP ile CrewAI çok aracılı Twitter araştırma ekipleri oluşturun.

TwexAPI’nin uzak MCP sunucusu aracılığıyla bir CrewAI MCP entegrasyonu oluşturun. CrewAI temsilcilerine kontrollü tweet aramaları, profil aramaları, takipçi aktarımları ve incelenen X eylemlerini verin. Her tweet kimliğini, profil kimliğini, imleci ve rota adını koruyun.

CrewAI’yi neden TwexAPI MCP ile kullanmalısınız?

CrewAI, karmaşık görevler için bir aracı çerçevesi sunar. Her temsilciye bir rol verin. TwexAPI, “explore” ve “twexapi_request” yoluyla uç nokta keşfi ve Twitter API işlemlerini sağlar.

Bu model araştırma, doğrulama ve raporlamaya uygundur. Model kararları gerektirmeyen deterministik işler için Python SDK veya doğrudan REST’i kullanın.

Önkoşullar

• Python 3.10’dan 3.13’e
• Bir TwexAPI API anahtarı
• CrewAI tarafından desteklenen bir LLM sağlayıcı anahtarı
• Yazma işlemleri için bir Twitter çerezi veya “auth_token”

Kurulum

CrewAI çekirdeği yerel bir MCP istemcisi içerir.

python -m pip install "crewai>=1.0" python-dotenv

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

export TWEXAPI_API_KEY="YOUR_API_KEY"
export OPENAI_API_KEY="YOUR_OPENAI_KEY"

Yazılı bir tweet arama ekibi oluşturun

Beklenen çıktıyla başlayın, ardından görevi oluşturun. CrewAI, Pydantic modeline göre son geçişi doğruluyor.

import os
from pathlib import Path
from typing import Literal

from crewai import Agent, Crew, Process, Task
from crewai.mcp import MCPServerHTTP
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
    pages_fetched: int
    stop_reason: Literal["complete", "requested_limit", "cursor_stalled"]


twexapi_mcp = MCPServerHTTP(
    url="https://api.twexapi.io/mcp",
    headers={"x-api-key": os.environ["TWEXAPI_API_KEY"]},
    streamable=True,
    cache_tools_list=True,
)

researcher = Agent(
    role="Twitter API Researcher",
    goal="Return exact tweet records and resumable pagination state",
    backstory=(
        "You inspect Twitter conversations through TwexAPI MCP. "
        "Call explore before twexapi_request. "
        "You preserve source IDs and never invent missing fields."
    ),
    llm="openai/gpt-4o",
    mcps=[twexapi_mcp],
    allow_delegation=False,
    verbose=False,
)

search_task = Task(
    description=(
        "Use TwexAPI MCP to search 50 latest tweets about CrewAI Twitter MCP. "
        "Call explore first. Use POST /twitter/advanced_search/page. "
        "Preserve exact tweet IDs, created timestamps, and cursors. "
        "Stop at the requested limit. Stop if a cursor repeats."
    ),
    expected_output="A validated tweet search handoff with pagination state.",
    agent=researcher,
    output_pydantic=TweetSearchHandoff,
)

crew = Crew(
    agents=[researcher],
    tasks=[search_task],
    process=Process.sequential,
    verbose=False,
)

result = crew.kickoff()
if result.has_tool_failures:
    raise RuntimeError("TwexAPI MCP tool failed. Inspect result.tool_failures.")

handoff = TweetSearchHandoff.model_validate(result.to_dict())
Path("twexapi-crewai-handoff.json").write_text(
    handoff.model_dump_json(indent=2),
    encoding="utf-8",
)

Her zaman has_tool_failuresı inceleyin. Eksik sonuçları hiçbir zaman yazma eylemlerine veya dışa aktarmalara aktarmayın.

Odaklanmış sorgularla tweet’leri arayın

İzleme için ‘sortBy: En Son’ seçeneğini kullanın. Etkileşim sıralamalı araştırma için “En İyi”yi kullanın. ’Sonraki_imleç’i değiştirmeden iletin.

Rol bazlı bir araştırma ekibi oluşturun

TwexAPI MCP’ye yalnızca araştırmacıya erişim izni verin. Doğrulanmış görevini araç kullanmadan analistlere aktarın.

class TweetAnalysis(BaseModel):
    query: str
    analyzed_tweet_ids: list[str]
    recurring_topics: list[str]
    top_author_usernames: list[str]
    next_cursor: str | None = None

analyst = Agent(
    role="Tweet Conversation Analyst",
    goal="Analyze only the supplied tweet rows",
    backstory="You compare exact tweets without fetching extra records.",
    llm="openai/gpt-4o",
    allow_delegation=False,
)

analysis_task = Task(
    description=(
        "Analyze the supplied tweet rows. "
        "Keep every analyzed tweet_id. Preserve the next_cursor."
    ),
    expected_output="A typed topic analysis tied to source tweet IDs.",
    agent=analyst,
    context=[search_task],
    output_pydantic=TweetAnalysis,
)

research_crew = Crew(
    agents=[researcher, analyst],
    tasks=[search_task, analysis_task],
    process=Process.sequential,
)

Yalnızca uç nokta keşfini kullanıma sunma

Uç nokta keşfi için yalnızca “keşfet”i gösterin.

from crewai.mcp import MCPServerHTTP
from crewai.mcp.filters import create_static_tool_filter

discovery_mcp = MCPServerHTTP(
    url="https://api.twexapi.io/mcp",
    headers={"x-api-key": os.environ["TWEXAPI_API_KEY"]},
    tool_filter=create_static_tool_filter(
        allowed_tool_names=["explore"],
    ),
    cache_tools_list=True,
)

‘Twexapi_request’ eklenmesi yetkili yürütmeyi etkinleştirir.

Twitter eylemlerini araştırma ekibinin dışında tutun

Otonom araştırma ekiplerine asla yazma izni vermeyin.

class TweetWritePlan(BaseModel):
    tweet_content: str
    reply_to_tweet_id: str | None = None
    media_urls: list[str]
    requires_human_confirmation: bool = True

planner = Agent(
    role="Twitter Action Planner",
    goal="Prepare one reviewable X action without executing it",
    backstory="You preserve approved text, target IDs, and route names.",
    llm="openai/gpt-4o",
    tools=[],
    allow_delegation=False,
)

plan_task = Task(
    description="Prepare a tweet or reply plan from reviewed source tweets.",
    expected_output="One typed action plan. Do not execute any X request.",
    agent=planner,
    output_pydantic=TweetWritePlan,
    human_input=True,
)

Onaylandıktan sonra bir REST veya SDK isteği gönderin. Önce CLI --dry-run ile yazma yüklerini önizleyin.

Hataları ve takım arızalarını ele alın

Arızalardan sonra result.tool_failuresı inceleyin. ’429’dan sonra ’next_cursor’u ve tamamlanmış tweet kimliklerini koruyun.

Paket versiyonları

Sonraki adımlar

• MCP Araçları
• Ajan MCP Aktarımı
• LangChain
• Gelişmiş Twitter Araması