Google ADK Integration

ArmorIQADK wraps Google's Agent Development Kit so every tool call produced by an ADK agent goes through ArmorIQ's plan / token / policy / audit pipeline. You keep your existing LlmAgent, McpToolset, and runner wiring - you just install the ArmorIQ scope on the agent for the duration of each request.

Live in Python and TypeScript. See Integrations for the full status matrix.

Install

pip install "armoriq-sdk[google-adk]"

npm install @armoriq/sdk
# plus your Google ADK agent package

Minimal example

from armoriq_sdk.integrations.google_adk import ArmorIQADK

# Once per process
armoriq = ArmorIQADK(api_key=os.environ["ARMORIQ_API_KEY"])

# Per-request
scope = armoriq.for_user("alice@example.com", goal=user_message)
scope.install(root_agent)
try:
    async for event in runner.run_async(...):
        handle(event)
finally:
    scope.uninstall(root_agent)

import { ArmorIQADK } from '@armoriq/sdk/dist/integrations/google_adk';

// Once per process
const armoriq = new ArmorIQADK({ apiKey: process.env.ARMORIQ_API_KEY! });

// Per-request
const scope = await armoriq.forUser('alice@example.com', { goal: userMessage });
scope.install(rootAgent);
try {
  for await (const event of runner.runAsync(...)) {
    handle(event);
  }
} finally {
  scope.uninstall(rootAgent);
}

That's it. The SDK installs three ADK lifecycle callbacks on your agent:

Callback	What ArmorIQ does
`after_model_callback`	Builds a plan from the LLM's tool calls and mints an intent token
`before_tool_callback`	Enforces the user's policy - allow, hold (wait for approval), or block
`after_tool_callback`	Audits the call with the result

Full example: an ADK agent with GitHub + Stripe MCPs

This mirrors the sdk-adk-test-agent-py reference app. It is Python, but the TypeScript API is the same shape: new ArmorIQADK({ apiKey }), then forUser, install, and uninstall.

import os
from dotenv import load_dotenv
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from google.adk.agents import LlmAgent
from google.adk.runners import InMemoryRunner
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset, StdioConnectionParams
from google.genai import types as genai_types
from mcp import StdioServerParameters

from armoriq_sdk.integrations.google_adk import ArmorIQADK

load_dotenv()

# ── Once per process ────────────────────────────────────────────────
armoriq = ArmorIQADK(api_key=os.environ["ARMORIQ_API_KEY"])

root_agent = LlmAgent(
    model=os.environ.get("ADK_MODEL", "gemini-2.5-flash"),
    name="demo_agent",
    description="Agent with GitHub + Stripe tools under ArmorIQ policy.",
    instruction="Use the tool that matches the user's request.",
    tools=[
        McpToolset(connection_params=StdioConnectionParams(
            server_params=StdioServerParameters(
                command="npx",
                args=["-y", "@modelcontextprotocol/server-github"],
                env={"GITHUB_PERSONAL_ACCESS_TOKEN": os.environ["GITHUB_TOKEN"]},
            ),
        )),
        McpToolset(connection_params=StdioConnectionParams(
            server_params=StdioServerParameters(
                command="npx",
                args=["-y", "@stripe/mcp", f"--api-key={os.environ['STRIPE_API_KEY']}"],
            ),
        )),
    ],
)

# ── Per request ─────────────────────────────────────────────────────
app = FastAPI()

@app.post("/chat/stream")
async def chat_stream(req: Request):
    body = await req.json()
    message = body["message"]
    user_email = body["userEmail"]

    async def events():
        scope = armoriq.for_user(user_email, goal=message)
        scope.install(root_agent)
        runner = InMemoryRunner(agent=root_agent, app_name="demo")
        try:
            session = await runner.session_service.create_session(
                app_name=runner.app_name, user_id=user_email
            )
            async for event in runner.run_async(
                user_id=user_email,
                session_id=session.id,
                new_message=genai_types.Content(
                    role="user", parts=[genai_types.Part(text=message)],
                ),
            ):
                yield f"data: {event}\n\n"
        finally:
            scope.uninstall(root_agent)

    return StreamingResponse(events(), media_type="text/event-stream")

What you need to add to an existing agent

If you already have an ADK agent, the diff is four lines:

+ from armoriq_sdk.integrations.google_adk import ArmorIQADK
+ armoriq = ArmorIQADK(api_key=os.environ["ARMORIQ_API_KEY"])

  async def handle_chat(user_email: str, message: str):
+     scope = armoriq.for_user(user_email, goal=message)
+     scope.install(root_agent)
      try:
          async for event in runner.run_async(...):
              ...
+     finally:
+         scope.uninstall(root_agent)

No changes to your tools, no changes to your LlmAgent definition, no plan-building by hand.

`ArmorIQADK` parameters

Parameter	Type	Default	Description
`api_key`	str	-	Your ArmorIQ API key
`backend_endpoint`	str	production	Override the control-plane URL
`iap_endpoint`	str	production	Override the IAP URL
`proxy_endpoint`	str	production	Override the proxy URL
`use_production`	bool	`True`	Set `False` for local dev endpoints
`default_mcp_name`	str	`None`	MCP name to attribute tools that can't be auto-mapped
`tool_name_parser`	callable	built-in	Custom `tool_name -> (mcp, action)` mapper
`validity_seconds`	int	`300`	Intent-token validity window
`mode`	str	`"sdk"`	Session mode (SDK / server / local)
`llm`	str	`"agent"`	Label recorded in the audit trail

`for_user(email, *, goal=None)`

Returns a per-request bundle with three methods:

Method	Description
`scope.install(agent)`	Attach the three ADK callbacks to the agent
`scope.uninstall(agent)`	Detach them - always run this in `finally`
`scope.session`	The underlying `ArmorIQSession` (for advanced use)

The bundle caches plan state for the request. Tool calls that aren't in the plan get blocked before they hit the MCP.

Testing

Use use_production=False + BACKEND_ENDPOINT=http://localhost:3000 to point at a local control plane.
Set DEFAULT_USER_EMAIL in your test harness so test runs don't hit your real users' policies.
Bootstrap once (armoriq.bootstrap()) and reuse - it's cached.

Reference implementation

Full source: sdk-adk-test-agent-py - a FastAPI + ADK + GitHub/Stripe MCP reference agent wired for ArmorIQ.

Google ADK

On this page