Microsoft Agent Framework Twitter MCP Python guide

python -m pip install "agent-framework==1.13.0"

import asyncio
import os
from pathlib import Path
from typing import Literal

from agent_framework import Agent, MCPStreamableHTTPTool
from agent_framework.openai import OpenAIChatClient
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",
    ]


async def main() -> None:
    xquik_mcp = MCPStreamableHTTPTool(
        name="xquik-twitter-api",
        url="https://xquik.com/mcp",
        description="Search tweets, profiles, followers, monitors, and exports.",
        allowed_tools=["explore", "xquik"],
        header_provider=lambda kwargs: {
            "x-api-key": kwargs["xquik_api_key"],
        },
    )

    async with Agent(
        client=OpenAIChatClient(model=os.environ["OPENAI_CHAT_MODEL"]),
        name="xquik_tweet_search_agent",
        instructions=(
            "Use GET /api/v1/x/tweets/search. "
            "Preserve exact tweet IDs, created timestamps, and cursors. "
            "Never invent missing tweet fields. "
            "Stop at the requested limit. Stop if a cursor repeats."
        ),
        tools=xquik_mcp,
    ) as agent:
        result = await agent.run(
            "Search 50 latest tweets about Microsoft Agent Framework MCP.",
            options={"response_format": TweetSearchHandoff},
            function_invocation_kwargs={
                "xquik_api_key": os.environ["XQUIK_READ_ONLY_API_KEY"],
            },
        )

        if not isinstance(result.value, TweetSearchHandoff):
            raise RuntimeError("Tweet handoff invalid. Inspect the agent response.")

        Path("xquik-microsoft-agent-handoff.json").write_text(
            result.value.model_dump_json(indent=2),
            encoding="utf-8",
        )


asyncio.run(main())

from agent_framework import Agent, MCPStreamableHTTPTool, Message

approved_mcp = MCPStreamableHTTPTool(
    name="xquik-twitter-api",
    url="https://xquik.com/mcp",
    allowed_tools=["explore", "xquik"],
    approval_mode={
        "never_require_approval": ["explore"],
        "always_require_approval": ["xquik"],
    },
    header_provider=lambda kwargs: {
        "x-api-key": kwargs["xquik_api_key"],
    },
)

async with Agent(
    client=OpenAIChatClient(model=os.environ["OPENAI_CHAT_MODEL"]),
    name="approved_xquik_agent",
    instructions="Show exact routes and arguments before every Xquik request.",
    tools=approved_mcp,
) as agent:
    session = agent.create_session()
    run_options = {"response_format": TweetSearchHandoff}
    run_context = {"xquik_api_key": os.environ["XQUIK_API_KEY"]}

    result = await agent.run(
        "Search tweets about Python Twitter MCP agents.",
        session=session,
        options=run_options,
        function_invocation_kwargs=run_context,
    )

    while result.user_input_requests:
        responses = []
        for request in result.user_input_requests:
            if request.function_call is None:
                responses.append(
                    request.to_function_approval_response(approved=False)
                )
                continue

            print("Function:", request.function_call.name)
            print("Arguments:", request.function_call.arguments)
            answer = await asyncio.to_thread(input, "Approve this call? [y/N] ")
            responses.append(
                request.to_function_approval_response(
                    approved=answer.strip().lower() == "y"
                )
            )

        result = await agent.run(
            Message(role="user", contents=responses),
            session=session,
            options=run_options,
            function_invocation_kwargs=run_context,
        )

discovery_mcp = MCPStreamableHTTPTool(
    name="xquik-route-discovery",
    url="https://xquik.com/mcp",
    allowed_tools=["explore"],
    header_provider=lambda kwargs: {
        "x-api-key": kwargs["xquik_api_key"],
    },
)

def tenant_headers(kwargs: dict[str, object]) -> dict[str, str]:
    tenant_id = str(kwargs["tenant_id"])
    return {"x-api-key": secret_store.get_xquik_key(tenant_id)}


tenant_mcp = MCPStreamableHTTPTool(
    name="tenant-xquik-twitter-api",
    url="https://xquik.com/mcp",
    allowed_tools=["explore", "xquik"],
    header_provider=tenant_headers,
)

async with Agent(
    client=OpenAIChatClient(model=os.environ["OPENAI_CHAT_MODEL"]),
    name="tenant_twitter_agent",
    instructions="Preserve exact profile IDs and usernames.",
    tools=tenant_mcp,
) as tenant_agent:
    result = await tenant_agent.run(
        "Look up the profile for microsoft.",
        function_invocation_kwargs={"tenant_id": authenticated_tenant_id},
    )

researcher = Agent(
    client=client,
    name="tweet_researcher",
    instructions="Return validated tweet rows and pagination state.",
    tools=xquik_mcp,
)

reviewer = Agent(
    client=client,
    name="tweet_reviewer",
    instructions="Analyze only supplied tweets. Preserve every tweet_id.",
)

research = await researcher.run(
    "Search latest tweets about Python agent frameworks.",
    options={"response_format": TweetSearchHandoff},
    function_invocation_kwargs={"xquik_api_key": tenant_api_key},
)

if not isinstance(research.value, TweetSearchHandoff):
    raise RuntimeError("Tweet research invalid. Stop the handoff.")

review = await reviewer.run(research.value.model_dump_json())