> ## Documentation Index
> Fetch the complete documentation index at: https://docs.hexgate.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Google ADK

> HexgateRunner — wraps google.adk.runners.Runner with policy + observability.

<Note>
  New to the adapters? Start with the [wrapping overview and comparison
  table](/adapters/overview#how-framework-wrapping-works).
</Note>

<Info>
  **Compatible versions:** see the [compatibility
  table](/adapters/overview#compatible-framework-versions) for the verified
  `google-adk` range.
</Info>

The Google ADK wrapper exposes its own `HexgateRunner`. It's constructed up front
with the agent, app name, and session service (mirroring the ADK `Runner`
constructor) — the underlying ADK `Runner` is built once and reused since role
resolution happens at call time. `run` / `run_async` then yield ADK events.

```python theme={null}
import asyncio

from dotenv import load_dotenv
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.sessions import InMemorySessionService
from google.genai import types

from hexgate.runtime import HexgateContext
from hexgate.adapters.google import HexgateRunner

INSTRUCTION = (
    "You are a DevOps assistant for a Kubernetes platform. Help engineers read "
    "service logs, restart services, and scale deployments. Pull the service "
    "name, replica count, and environment (dev/staging/prod) from the request "
    "and act directly — the policy layer gates sensitive actions."
)


def read_logs(service: str, env: str) -> str:
    """Return recent log lines for `service` in `env` (dev/staging/prod)."""
    return f"(stub) {service}@{env}: 200 OK, 200 OK, WARN upstream slow"


def restart_service(service: str, env: str) -> str:
    """Restart `service` in `env`."""
    return f"(stub) restarted {service}@{env}"


def scale_deployment(service: str, replicas: int, env: str) -> str:
    """Scale `service` to `replicas` pods in `env`."""
    return f"(stub) scaled {service}@{env} to {replicas} replicas"


async def main():
    load_dotenv()

    agent = Agent(
        name="devops_agent",
        model=LiteLlm(model="openai/gpt-4o"),
        instruction=INSTRUCTION,
        tools=[read_logs, restart_service, scale_deployment],
    )

    hexgate_context = HexgateContext(user_id="engineer_1", session_id="session_1", user_roles=["operator"])

    session_service = InMemorySessionService()
    await session_service.create_session(
        app_name="devops_demo",
        user_id=hexgate_context.user_id,
        session_id=hexgate_context.session_id,
    )

    runner = HexgateRunner(
        agent=agent,
        app_name="devops_demo",
        session_service=session_service,
    )  # picks up HEXGATE_API_KEY from env

    user_msg = types.Content(
        role="user",
        parts=[types.Part(text="Scale the checkout service to 5 replicas in staging.")],
    )

    async for event in runner.run_async(new_message=user_msg, hexgate_context=hexgate_context):
        if event.is_final_response():
            print(event.content.parts[0].text)


if __name__ == "__main__":
    asyncio.run(main())
```

## Under the hood

* At construction, `HexgateRunner` calls `wrap_google_agent`, which builds a
  `PolicySet`, constructs one `PolicyEnforcer`, and returns
  `agent.model_copy(update={"tools": guarded_tools})` — your original `agent` is
  untouched.
* Each tool is normalized first: bare callables in `agent.tools` are wrapped into
  `FunctionTool` (matching what ADK does internally) so the guard has a stable
  `BaseTool` surface. Each tool is then `copy.copy`'d and its `run_async` replaced
  with an enforcer-gated version.
* Each `run` / `run_async` call opens a `HexgateContext` scope (`hexgate_context.sync_scope()` /
  `async with hexgate_context:`) and dispatches to the cached underlying `Runner`. On
  non-allow, the guard returns `decision.as_error_message()` so the ADK runtime
  forwards it to the model as the tool output instead of aborting the run.
* Observability is set up lazily on each call: `GoogleADKInstrumentor().instrument()`
  plus `nest_asyncio.apply()` (ADK's runner spins its own loop), and the run
  executes inside `propagate_attributes(user_id=..., session_id=...,
  metadata={"user_roles": ...}, tags=["google.runner.run.<agent_name>"])`.

## Runnable example

`examples/devops_google.py` — `HexgateRunner` (Google ADK) end-to-end with
`InMemorySessionService`.