> ## Documentation Index
> Fetch the complete documentation index at: https://docs.xquik.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Google ADK
> Build a multi-agent Twitter assistant with Google ADK and Xquik's MCP tools
For the complete documentation index, see llms.txt.
Build a Google ADK agent that can search tweets, hand off IDs and cursors, post tweets, replay stored monitor events, and run extraction jobs - powered by Xquik's MCP server and Gemini.
## Prerequisites
* Python 3.10+
* [Xquik API key](/quickstart) (`xq_...`)
* A Google AI API key (for Gemini models)
## Install
```bash theme={null}
pip install google-adk
```
## Full Example
```python theme={null}
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())
```
The MCP runtime returns normalized snake\_case fields through `xquik.request()`, so keep ADK prompts aligned with `tweet_id`, `has_more`, `next_cursor`, and returned job IDs.
## Handoff Checklist
Store `tweet_id`, `text`, `author_username`, `created_at`, `has_more`, `next_cursor`, and the original `q`.
Store source `id` as `user_id`, plus `username`, `name`, `followers`, `verified`, `profile_picture`, `has_more`, `next_cursor`, and the source lookup or search query.
Store each trend `name`, `rank`, `query`, and `description`. Keep response `count`, `woeid`, and the requested region with the run checkpoint.
Store the returned monitor `id` as `monitor_id`, `event_types`, `next_billing_at`, the returned webhook `id` as `webhook_id`, `url`, and the one-time `secret` in a secret manager. On production deliveries, store `delivery_id` for receiver retry de-dupe and `stream_event_id` when one monitor event should process once across endpoint changes.
Store `event_id`, `type`, `monitor_id`, `monitor_type`, `occurred_at`, `has_more`, `next_cursor`, and the `after` query for the next page.
Store `extraction_id`, `status`, `poll`, and `export_after_complete`; poll before loading CSV, JSON, or XLSX rows.
Store `tweet_id` or `write_action_id`, `reply_to_tweet_id`, `status`, `charged_credits`, and `poll`; do not resend pending writes.
For tweets or replies, pass public URLs in `media` and store `tweet_id` or `write_action_id`. For DMs, upload first, pass one `media_id` in `media_ids`, store `message_id`, and leave `reply_to_message_id` unset.
## Multi-Agent Setup
ADK supports hierarchical agents. Use a root agent that delegates to specialized sub-agents:
```python theme={null}
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],
)
```
## Dynamic Headers
Use `header_provider` for per-request headers (e.g., multi-tenant apps with per-user API keys):
```python theme={null}
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,
)
```
## Tool Filtering
Expose only specific tools to the agent:
```python theme={null}
xquik_toolset = McpToolset(
connection_params=StreamableHTTPConnectionParams(
url="https://xquik.com/mcp",
headers={"x-api-key": "xq_YOUR_KEY_HERE"},
),
tool_filter=["explore"], # Read-only: only the explore tool
)
```
## Environment Variables
```bash .env theme={null}
XQUIK_API_KEY=xq_YOUR_KEY_HERE
GOOGLE_API_KEY=...
```
```python theme={null}
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"]},
),
)
```
Use `McpToolset` (lowercase 'c'), not `MCPToolset`. The uppercase version is deprecated and emits a warning.
## Package Versions
| Package | Version |
| ------------ | ------- |
| `google-adk` | 1.28.1+ |
| `mcp` | 1.23.0+ |