Google ADK

pip install "google-adk[mcp]"

import asyncio
from pathlib import Path
from google.adk.agents import LlmAgent
from google.adk.runners import InMemoryRunner
from google.adk.tools.mcp_tool import McpToolset, StreamableHTTPConnectionParams
from google.genai import types


async def main():
    xquik_toolset = McpToolset(
        connection_params=StreamableHTTPConnectionParams(
            url="https://xquik.com/mcp",
            headers={"x-api-key": "xq_YOUR_KEY_HERE"},
        ),
    )

    agent = LlmAgent(
        model="gemini-2.5-flash",
        name="xquik_agent",
        instruction="You help users interact with X (Twitter) via the Xquik API.",
        tools=[xquik_toolset],
    )

    runner = InMemoryRunner(agent=agent, app_name="xquik_app")
    session = await runner.session_service.create_session(
        app_name="xquik_app",
        user_id="user-1",
    )

    handoff_prompt = (
        "Search for tweets about AI agents. Return compact JSON with "
        "query, route_used, tweets[{tweet_id,text,author_username,created_at}], "
        "has_more, next_cursor."
    )
    response_parts = []

    async for event in runner.run_async(
        user_id="user-1",
        session_id=session.id,
        new_message=types.Content(
            role="user",
            parts=[types.Part(text=handoff_prompt)],
        ),
    ):
        if event.content and event.content.parts:
            for part in event.content.parts:
                if part.text:
                    response_parts.append(part.text)

    Path("xquik-adk-handoff.json").write_text(
        "".join(response_parts),
        encoding="utf-8",
    )

    await xquik_toolset.close()


asyncio.run(main())

from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import McpToolset, StreamableHTTPConnectionParams

xquik_toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://xquik.com/mcp",
        headers={"x-api-key": "xq_YOUR_KEY_HERE"},
    ),
)

researcher = LlmAgent(
    model="gemini-2.5-flash",
    name="researcher",
    instruction=(
        "Search X for tweets and user profiles. Return compact JSON with "
        "tweet_id, author_username, text, created_at, has_more, next_cursor, "
        "and route_used."
    ),
    tools=[xquik_toolset],
)

analyst = LlmAgent(
    model="gemini-2.5-flash",
    name="analyst",
    instruction="Analyze tweet data and identify trends, sentiment, and key influencers.",
)

coordinator = LlmAgent(
    model="gemini-2.5-flash",
    name="coordinator",
    instruction="""You coordinate research tasks about X (Twitter).
    Delegate data collection to the researcher agent and analysis to the analyst agent.""",
    sub_agents=[researcher, analyst],
)

from google.adk.tools.mcp_tool import McpToolset, StreamableHTTPConnectionParams

def get_headers(context):
    # Access session state for per-user API keys
    return {"x-api-key": context.state.get("xquik_api_key", "")}

xquik_toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://xquik.com/mcp",
    ),
    header_provider=get_headers,
)

xquik_toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://xquik.com/mcp",
        headers={"x-api-key": "xq_YOUR_KEY_HERE"},
    ),
    tool_filter=["explore"],  # Discovery only: cannot execute API operations
)

import os
from google.adk.tools.mcp_tool import McpToolset, StreamableHTTPConnectionParams

xquik_toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://xquik.com/mcp",
        headers={"x-api-key": os.environ["XQUIK_API_KEY"]},
    ),
)