Google ADK

Secure your Google ADK agent with ArmorIQ in three lines of glue code. Per-user scoping, per-tool policy enforcement, async-native, and delegation-aware.

The ArmorIQ SDK ships a first-class integration for Google Agent Development Kit (ADK). One factory lives for the life of your process; per-request, you build a scoped bundle tied to one end-user email and install it on your agent. ArmorIQ hooks into ADK's native before_tool_callback / after_tool_callback / after_model_callback lifecycle, so every tool call goes through plan → token → invoke automatically.

pip install armoriq-sdk google-adk

Setup in three lines

Create one factory at process start, then per request build a user-scoped bundle, install it, and uninstall after the run.

import os
from google.adk.agents import Agent
from armoriq_sdk.integrations.google_adk import ArmorIQADK

# 1. One factory per process — caches bootstrap (org, MCPs, toolMap).
armoriq = ArmorIQADK(api_key=os.environ["ARMORIQ_API_KEY"])

# Your regular ADK agent — nothing ArmorIQ-specific here.
root_agent = Agent(
    name="assistant",
    model="gemini-2.0-flash",
    tools=[...],
)

async def handle_request(user_email: str, message: str):
    # 2. Per-request bundle scoped to this end user.
    scope = armoriq.for_user(user_email, goal=message)

    # 3. Install ArmorIQ callbacks on the agent for this run.
    scope.install(root_agent)
    try:
        async for event in runner.run_async(user_id=user_email, new_message=message):
            yield event
    finally:
        # Always uninstall so the next request starts clean.
        scope.uninstall(root_agent)

What install() wires up, as ADK callbacks on the agent:

ADK hook	ArmorIQ work
`after_model_callback`	Capture the LLM's tool calls as a plan, mint an intent token
`before_tool_callback`	Check the policy for this user + tool; block, allow, or hold
`after_tool_callback`	Write the audit log for this invocation

Factory options

armoriq = ArmorIQADK(
    api_key=os.environ["ARMORIQ_API_KEY"],  # required
    default_mcp_name="travel",              # if a tool name has no mcp prefix
    tool_name_parser=None,                  # custom `(tool_name) → (mcp, action)` parser
    validity_seconds=300,                   # intent token lifetime
    mode="sdk",                             # session mode
    llm="agent",                            # label used in plan capture
)

Most integrations only need api_key.

How tool names map to MCPs

ADK tools have flat names (book_flight); ArmorIQ needs to know which MCP that tool lives on (travel-mcp.book_flight). The factory resolves this in order:

Platform mapping — ArmorIQ bootstraps your toolMap from the registered MCPs and finds the right MCP for each tool name.
Double-underscore convention — if your tool is named travel__book_flight, the factory splits on __ → (travel, book_flight).
default_mcp_name — fallback when the name has no other signal.
Custom tool_name_parser — pass a function (str) → (mcp, action) to override entirely.

Per-user scoping with `for_user()`

Every ADK run is tied to one end-user. Build the scope once per request and the bundle automatically tags every downstream plan, token, invocation, and audit log with that email.

scope = armoriq.for_user("alice@customer.com", goal="Plan a trip to Paris")

Per-user context (membership, policies, approver chain) is cached 5 minutes — calling for_user in a tight loop is cheap.

To invalidate the cache after a policy change:

armoriq.invalidate_user("alice@customer.com")

Multi-user SaaS pattern

A typical request handler looks like this. One factory, many user scopes:

from fastapi import FastAPI

app = FastAPI()
armoriq = ArmorIQADK(api_key=os.environ["ARMORIQ_API_KEY"])

@app.post("/chat")
async def chat(req: ChatRequest):
    scope = armoriq.for_user(req.user_email, goal=req.message)
    scope.install(root_agent)
    try:
        events = []
        async for event in runner.run_async(
            user_id=req.user_email,
            new_message=req.message,
        ):
            events.append(event)
        return {"events": events}
    finally:
        scope.uninstall(root_agent)

Every user's audit log is cleanly separated. Every policy that targets alice@customer.com (by email, by membership, or by role) is evaluated correctly.

Multi-agent hierarchies

ADK supports nested agents (agent-of-agents). The ArmorIQ bundle only installs callbacks on the agent you pass to install(). For nested agents, install on each one you want protected:

scope = armoriq.for_user(user_email)
scope.install(root_agent)
scope.install(child_agent_1)
scope.install(child_agent_2)
try:
    # ... run ...
finally:
    scope.uninstall(root_agent)
    scope.uninstall(child_agent_1)
    scope.uninstall(child_agent_2)

Policy holds and delegation

When a tool call requires human approval, ArmorIQ's before_tool_callback returns a hold — ADK blocks the tool call and surfaces the delegation info. Your run handler can wait for approval and resume.

The ADK bundle handles the hold/retry loop automatically when the policy is configured for auto-polling. Configure the hold-approval timeout on the factory:

armoriq = ArmorIQADK(
    api_key=os.environ["ARMORIQ_API_KEY"],
    validity_seconds=600,  # longer window for approval-gated flows
)

For the full delegation workflow (creating approvers, approving from a manager account, auditing decisions), see Policies and Delegation.

Troubleshooting

What you see	Why	Fix
`before_tool_callback failed` in logs	ArmorIQ couldn't reach the backend/proxy	Check network + `ARMORIQ_API_KEY`; calls fall through without enforcement on failure
Every tool blocked with `PolicyBlockedException`	Policy doesn't allow this user for these tools	Update the policy in the dashboard
Tool name resolves to `unknown.book_flight`	The factory can't map your tool name to an MCP	Set `default_mcp_name=...` or pass a custom `tool_name_parser`
`IntentMismatchException` after tool fires	ADK passed params that differ from what was captured in the plan	Usually means the tool's args aren't deterministic; make sure the tool's `params` match what the LLM reasoned about

Full worked example

import os
import asyncio
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from armoriq_sdk.integrations.google_adk import ArmorIQADK

# Your tools — same as vanilla ADK
def get_weather(city: str) -> dict:
    """Get current weather for a city."""
    return {"city": city, "temp": 22, "units": "celsius"}

# ArmorIQ factory — once per process
armoriq = ArmorIQADK(
    api_key=os.environ["ARMORIQ_API_KEY"],
    default_mcp_name="weather-mcp",
)

# ADK agent — nothing ArmorIQ-specific
root_agent = Agent(
    name="weather_bot",
    model="gemini-2.0-flash",
    description="A weather assistant",
    tools=[get_weather],
)

session_service = InMemorySessionService()
runner = Runner(agent=root_agent, app_name="weather-app", session_service=session_service)

async def ask(user_email: str, question: str):
    # Ensure a session exists for this user
    session = await session_service.create_session(
        app_name="weather-app",
        user_id=user_email,
        session_id="default",
    )

    # Wrap the run with ArmorIQ per-user scope
    scope = armoriq.for_user(user_email, goal=question)
    scope.install(root_agent)
    try:
        async for event in runner.run_async(
            user_id=user_email,
            session_id=session.id,
            new_message=types.Content(
                role="user",
                parts=[types.Part(text=question)],
            ),
        ):
            if event.content and event.content.parts:
                print(event.content.parts[0].text)
    finally:
        scope.uninstall(root_agent)

asyncio.run(ask("alice@customer.com", "What's the weather in Paris?"))

The call to get_weather goes through ArmorIQ's policy check — weather-mcp.get_weather for alice@customer.com — and the audit log shows the full chain.

Setup in three lines

Factory options

How tool names map to MCPs

Per-user scoping with `for_user()`

Multi-user SaaS pattern

Multi-agent hierarchies

Policy holds and delegation

Troubleshooting

Full worked example

Next steps

Custom frameworks

Policies

Python SDK

On this page