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.

Python only today. The TypeScript ADK adapter is coming soon. Every other framework adapter (CrewAI, LangChain, OpenAI, Anthropic) is also coming soon on both languages — see Integrations for the full status matrix.

Install

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

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)

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.

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