Google ADK
Connect Unizo MCP to Google ADK for building sophisticated AI agents with Gemini models and native MCP support.
Overview
Google's Agent Development Kit (ADK) is an open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents. ADK includes native MCP support via the McpToolset class, enabling seamless integration with Unizo's MCP server.
Installation
pip install google-adk
Or with uv:
uv add google-adk
Quick Start
Connect to Unizo MCP and create an ADK agent using the Runner pattern:
import os
import base64
import asyncio
from google.adk import Runner
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import McpToolset, StreamableHTTPConnectionParams
from google.adk.sessions import InMemorySessionService
from google.genai import types
# Configuration - API keys from environment (secrets)
UNIZO_API_KEY = os.getenv("UNIZO_API_KEY")
os.environ["GOOGLE_API_KEY"] = os.getenv("GOOGLE_API_KEY")
# Encode API key for Basic auth
auth_token = base64.b64encode(f"{UNIZO_API_KEY}:".encode()).decode()
def create_unizo_agent(account_id: str) -> LlmAgent:
"""
Create an agent for a specific Unizo account.
In production, account_id should come from:
- User/tenant context
- Database lookup based on authenticated user
- Request parameters
"""
return LlmAgent(
model="gemini-2.0-flash-exp",
name="unizo_assistant",
instruction="You are a helpful assistant with access to security, IT, and business data from connected platforms via Unizo.",
tools=[
McpToolset(
connection_params=StreamableHTTPConnectionParams(
url="https://api.unizo.ai/mcp",
headers={
"Authorization": f"Bearer {UNIZO_API_KEY}",
"x-mcp-scopes": "edr,vms,ticketing,scm"
}
),
tool_name_prefix="unizo_"
)
],
)
async def main():
# Get account_id from user context (not environment variable)
account_id = get_user_account_id() # Your function to determine this
# Create agent for this specific account
agent = create_unizo_agent(account_id)
# Create runner with in-memory session service
runner = Runner(
agent=agent,
app_name="unizo_demo",
session_service=InMemorySessionService()
)
# Create a session
await runner.session_service.create_session(
app_name="unizo_demo",
user_id="user_1",
session_id="session_1"
)
# Create and send a message
message = types.Content(parts=[types.Part(text="Check for critical vulnerabilities in our infrastructure")], role="user")
# Run the agent and collect response
async for event in runner.run_async(
user_id="user_1",
session_id="session_1",
new_message=message
):
if hasattr(event, 'content') and event.content:
if hasattr(event.content, 'parts'):
for part in event.content.parts:
if hasattr(part, 'text') and part.text:
print(part.text)
asyncio.run(main())
Environment Variables
export UNIZO_API_KEY="your_unizo_api_key"
export GOOGLE_API_KEY="your_google_api_key"
Account ID: The Unizo account ID should be determined from user context (e.g., which tenant/customer is using the agent), not from environment variables. Pass it as a parameter when creating the agent.
Multi-Turn Conversations
Continue conversations by reusing the same session:
# First query
message1 = types.Content(parts=[types.Part(text="List security incidents from SentinelOne")], role="user")
async for event in runner.run_async(user_id="user_1", session_id="session_1", new_message=message1):
# Process first response...
pass
# Follow-up query (uses same session for context)
message2 = types.Content(parts=[types.Part(text="Create tickets for any critical findings")], role="user")
async for event in runner.run_async(user_id="user_1", session_id="session_1", new_message=message2):
# Process follow-up response...
pass
Tool Filtering
Restrict which Unizo tools are exposed to the agent for security:
McpToolset(
connection_params=StreamableHTTPConnectionParams(
url="https://api.unizo.ai/mcp",
headers={
"Authorization": f"Bearer {UNIZO_API_KEY}",
"x-mcp-scopes": "edr,vms" # Only security tools
}
),
tool_filter=[
"edr_list_incidents",
"edr_get_incident_details",
"vms_list_vulnerabilities"
] # Only expose specific tools
)
Advanced Usage
Multi-Service Agent
Create an agent that can access multiple Unizo service categories:
def create_security_agent(account_id: str) -> LlmAgent:
"""Security operations agent with comprehensive toolset"""
return LlmAgent(
model="gemini-2.0-flash-exp",
name="security_ops_agent",
instruction="""You are a security operations assistant with access to:
- Endpoint Detection & Response (EDR) systems
- Vulnerability Management systems
- Ticketing systems for incident management
- Source Code Management for security reviews
Help with threat detection, incident response, and security monitoring.""",
tools=[
McpToolset(
connection_params=StreamableHTTPConnectionParams(
url="https://api.unizo.ai/mcp",
headers={
"Authorization": f"Bearer {UNIZO_API_KEY}",
"x-mcp-scopes": "edr,vms,ticketing,scm"
}
),
tool_name_prefix="unizo_"
)
],
)
Production Session Management
For production deployments, use persistent session storage:
from google.adk.sessions import PostgreSQLSessionService
# Production session service
session_service = PostgreSQLSessionService(
connection_string=os.getenv("DATABASE_URL")
)
runner = Runner(
agent=agent,
app_name="unizo_production",
session_service=session_service
)
Required Headers: Only Authorization and x-mcp-scopes are required. Additional headers like Content-Type, Accept, and MCP-Protocol-Version are handled automatically by the ADK client.
Class Name Update: In ADK v1.17.0+, use McpToolset (not MCPToolset). The old capitalization is deprecated but still works with a warning.
Available Scopes
Configure x-mcp-scopes to control which Unizo services your agent can access:
edr: Endpoint Detection & Response toolsvms: Vulnerability Management toolsticketing: Incident and ticket management toolsscm: Source Code Management toolscommunications: Team communication toolsobservability: Monitoring and logging toolsinfra: Infrastructure management tools