CrewAI Twitter MCP multi-agent guide for Python

python -m pip install "crewai>=1.15,<1.16"

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
    created: int | None
    url: str | None


class TweetSearchHandoff(BaseModel):
    query: str
    route_used: Literal["GET /api/v1/x/tweets/search"]
    tweets: list[TweetRow]
    has_more: bool
    next_cursor: str | None
    pages_fetched: int
    stop_reason: Literal[
        "complete",
        "requested_limit",
        "cursor_stalled",
    ]


xquik_mcp = MCPServerHTTP(
    url="https://xquik.com/mcp",
    headers={"x-api-key": os.environ["XQUIK_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 documented Xquik routes. "
        "You preserve source IDs and never invent missing fields."
    ),
    llm="openai/gpt-5",
    mcps=[xquik_mcp],
    allow_delegation=False,
    verbose=False,
)

search_task = Task(
    description=(
        "Use GET /api/v1/x/tweets/search. "
        "Search 50 latest tweets about CrewAI Twitter MCP. "
        "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("Twitter MCP tool failed. Inspect result.tool_failures.")

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

class TweetAnalysis(BaseModel):
    query: str
    analyzed_tweet_ids: list[str]
    recurring_topics: list[str]
    top_author_usernames: list[str]
    next_cursor: str | 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-5",
    allow_delegation=False,
    verbose=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,
    verbose=False,
)

class TweetWritePlan(BaseModel):
    account_id: str
    text: str
    reply_to_tweet_id: str | None
    media_urls: list[str]
    idempotency_key: str


planner = Agent(
    role="Twitter Action Planner",
    goal="Prepare one reviewable X action without executing it",
    backstory="You preserve approved text, target IDs, and account IDs.",
    llm="openai/gpt-5",
    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,
)

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


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

def build_tenant_mcp(user_api_key: str) -> MCPServerHTTP:
    return MCPServerHTTP(
        url="https://xquik.com/mcp",
        headers={"x-api-key": user_api_key},
        streamable=True,
        cache_tools_list=True,
    )