CrewAI

pip install crewai python-dotenv

import os
from pathlib import Path

from crewai import Agent, Crew, Process, Task
from crewai.mcp import MCPServerHTTP
from dotenv import load_dotenv

load_dotenv()

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

researcher = Agent(
    role="X Research Analyst",
    goal="Collect structured public X/Twitter data with Twexapi MCP",
    backstory="A careful analyst who preserves IDs, cursors, and route names.",
    mcps=[twexapi_mcp],
)

task = Task(
    description=(
        "Search recent tweets about AI agents. Use explore first. Return compact JSON "
        "with route_used, tweets[{tweet_id,text,author_username,created_at}], "
        "has_more, next_cursor, and key influencers."
    ),
    expected_output="Compact JSON with tweet rows, pagination state, and influencer notes.",
    agent=researcher,
)

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

result = crew.kickoff()
Path("twexapi-crewai-handoff.json").write_text(str(result), encoding="utf-8")

analyst = Agent(
    role="Engagement Analyst",
    goal="Find patterns in tweet rows and profile rows",
    backstory="Transforms structured social data into concise findings.",
)

writer = Agent(
    role="Report Writer",
    goal="Write a short executive summary from the analyst output",
    backstory="Creates readable summaries without inventing missing metrics.",
)

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

read_only_catalog = 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"],
    ),
)

import os
from crewai.mcp import MCPServerHTTP

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