auth-framework-config

auth-framework-config

Configure authentication and authorization for an ADK 2.0 agent and its tools.

Three layers

1. Endpoint auth ──▶ Who can talk to the agent at all
2. Agent auth   ──▶ User identity propagated to tools
3. Tool auth    ──▶ Scopes required per tool

Layer 1: endpoint auth (Cloud Run + IAP)

gcloud run services add-iam-policy-binding my-agent \
    --member="user:[email protected]" \
    --role="roles/run.invoker"

Or for public + token check:

from fastapi import Header, HTTPException

@app.middleware("http")
async def verify_token(request, call_next):
    token = request.headers.get("authorization", "").removeprefix("Bearer ")
    if not verify_jwt(token):
        raise HTTPException(401, "Invalid token")
    return await call_next(request)

Layer 2: propagating user identity

from google.adk.runners import Runner

@app.post("/chat")
async def chat(req: ChatRequest, authorization: str = Header(...)):
    user_id = decode_jwt(authorization.removeprefix("Bearer "))["sub"]
    session = await session_service.create_session(
        app_name="my_agent",
        user_id=user_id,
        state={"auth_token": authorization, "principal": user_id},
    )
    response = await runner.run_async(session=session, input=req.message)
    return response

Layer 3: per-tool auth scopes

from google.adk.tools.openapi_tool.auth import AuthScheme, AuthCredential

# OAuth2 scheme for a Google API tool
gmail_scheme = AuthScheme(
    type="oauth2",
    flows={
        "authorizationCode": {
            "scopes": ["https://www.googleapis.com/auth/gmail.readonly"]
        }
    },
)
gmail_cred = AuthCredential(
    oauth2={"client_id_env": "GMAIL_CLIENT_ID", "client_secret_env": "GMAIL_SECRET"},
)

gmail_toolset = OpenAPIToolset(
    spec_url="https://gmail.googleapis.com/$discovery/rest?version=v1",
    auth_scheme=gmail_scheme,
    auth_credential=gmail_cred,
)

Tool-level scope check (custom tool)

from google.adk.tools import ToolContext

async def delete_record(record_id: str, ctx: ToolContext) -> dict:
    """Delete a record. Requires 'records.delete' scope."""
    user_scopes = ctx.session.state.get("scopes", [])
    if "records.delete" not in user_scopes:
        return {"error": "permission_denied", "required_scope": "records.delete"}
    db.delete(record_id)
    return {"deleted": True}

API key auth (simpler)

import os
from fastapi import Header, HTTPException

API_KEY = os.environ["AGENT_API_KEY"]

@app.middleware("http")
async def api_key_check(request, call_next):
    if request.headers.get("x-api-key") != API_KEY:
        raise HTTPException(401)
    return await call_next(request)

Combining with HITL

Layer 3 (tool scopes) + tool-confirmation = belt-and-suspenders for sensitive tools:

from google.adk.tools import require_confirmation
guarded_delete = require_confirmation(delete_record)

Validation

• Endpoint rejects unauthenticated requests (401)
• Invalid tokens rejected (test with expired / wrong-aud)
• Tool scope checks fire (test with under-scoped user)
• Audit logs capture the principal on every call

See also

• safety-policy-enforcer for content-level controls
• tool-confirmation-hitl for approval-gate patterns