🔒 HTTP Bearer in ADK

[PanchoG17]

Example Scenario: Using JWT in ADK-Python for Authenticated User CRUD

Hi community,

I’d like to discuss a basic scenario and get guidance on securely handling JWT tokens within the agent flow.

Scenario

Suppose we have a tool that performs basic CRUD operations on user accounts. Only authenticated users can modify their own data.

We want to use JWT tokens for authentication. The questions are:

1. How to securely extract the user_id from the JWT within the agent flow.
2. How to ensure that operations like update or delete are restricted to the authenticated user.
3. Any recommended best practices or helper functions in ADK-Python to make this process secure and maintainable.

Example Flow

1. User logs in and receives a JWT token (signed by our auth server).
2. User sends requests to the agent, including the token in the Authorization: Bearer <token> header.
3. Agent extracts the user_id from the token.
4. Agent performs CRUD operations only if the user_id matches the resource owner.

I’d love to see an example or discussion about the recommended way to handle this in ADK-Python, especially how to extract the JWT payload safely and integrate it into the flow of the agent.

Thanks for any guidance or examples!

[adk-bot]

Response from ADK Answering Agent (experimental, answer may be inaccurate)

TLDR: Use the "User Auth" pattern with an HTTPBearer authentication scheme. Your tool should request the JWT from the tool_context, and then pass this token to your backend API. The backend API is responsible for validating the token and enforcing user-specific permissions.

---

Hi PanchoG17,

Thanks for the great question. This is a common and important scenario for building secure agents. Here's a breakdown of the recommended way to handle JWT-based authentication in ADK-Python, addressing your specific points.

1. How to securely extract the user_id from the JWT

The ADK framework provides a "User Auth" pattern that is designed for this use case. [1] In this model, the agent acts on behalf of the end-user by using the user's credentials.

Here is the flow:

1. Define the Authentication Scheme: In your tool's configuration, you specify that it requires an HTTPBearer token. ADK will handle the process of requesting this token from the client application when the tool is called. [2]

2. Request the Credential in Your Tool: Inside your tool's implementation, you can access the user's provided token via the tool_context. You would use tool_context.request_credential(AUTH_CONFIG) to get the AuthCredential object containing the token. [3]

3. Decode the Token: Once you have the JWT string, you can decode it using a standard Python library like PyJWT. This will give you access to the claims inside, such as user_id.

Important: While your tool can decode the token to extract information, the validation of the token's signature should always be performed by your backend service to ensure it hasn't been tampered with.

2. How to ensure operations are restricted to the authenticated user

This is the core of the "User Auth" model. Instead of the agent trying to enforce permissions, it passes the responsibility to your existing backend API.

• Your tool, now in possession of the user's JWT, should include it in the Authorization header of any calls it makes to your backend CRUD API.
• Your backend API should then be responsible for:
  1. Validating the JWT (checking the signature, expiration, etc.).
  2. Extracting the user_id from the validated token.
  3. Enforcing business logic, i.e., checking that the user_id from the token is authorized to perform the requested action (e.g., update or delete a specific user's data).

This way, your security logic remains centralized in your backend, and the agent simply acts as a secure intermediary.

3. Recommended best practices or helper functions in ADK-Python

• Use adk.auth: Leverage the adk.auth module, specifically HTTPBearer for defining your authentication scheme. [2]
• Use ToolContext: The tool_context is the secure bridge for accessing session-specific information like credentials and state. [4, 5] Avoid trying to access headers or request data through other means.
• Keep Tools Focused: Design your tools to be specific to a task. The tool's responsibility is to get the credential, call the backend, and return the result. It should not contain complex business logic. [1]
• State Management: For passing data between tool calls within a single agent execution, you can use tool_context.state. [3, 4] However, be cautious about storing sensitive information like raw tokens in the state, especially if the state is persisted long-term.

Here is a conceptual code snippet of what a tool might look like:

from google.adk.tools import tool
from google.adk.tools.tool_context import ToolContext
import jwt # Example library

# Define your auth configuration, which would be loaded by the agent
AUTH_CONFIG = {
    "type": "HTTPBearer"
}

@tool
def update_user_profile(tool_context: ToolContext, user_id_to_update: str, new_data: dict) -> str:
    """Updates a user's profile."""

    # 1. Request the credential from the context
    auth_credential = tool_context.request_credential(AUTH_CONFIG)
    jwt_token = auth_credential.token

    # 2. (Optional) Decode to get user_id for logging or simple checks
    try:
        decoded_token = jwt.decode(jwt_token, options={"verify_signature": False})
        calling_user_id = decoded_token.get("sub") # 'sub' is a standard claim for user ID
    except jwt.DecodeError:
        return "Error: Invalid token format."

    # 3. Pass the original token to your backend for secure validation and authorization
    headers = {"Authorization": f"Bearer {jwt_token}"}
    
    # Your backend API would perform the check:
    # is calling_user_id allowed to update user_id_to_update?
    response = requests.put(
        f"https://your-api.com/users/{user_id_to_update}",
        json=new_data,
        headers=headers
    )

    if response.status_code == 200:
        return "User profile updated successfully."
    elif response.status_code == 403:
        return "Error: You are not authorized to perform this action."
    else:
        return f"Error: API returned status {response.status_code}."

By following this pattern, you can integrate securely with your existing authentication and authorization systems.

---

[1] https://google.github.io/adk-docs/safety/
[2] https://google.github.io/adk-docs/tools/authentication/
[3] https://google.github.io/adk-docs/tutorials/agent-team/
[4] https://google.github.io/adk-docs/context/
[5] https://github.com/google/adk-python/blob/main/src/google/adk/auth/auth_credential.py

[PanchoG17]

AUTH_CONFIG = {
    "type": "HTTPBearer"
}

is not a valid auth config. still not working

[Pittimon]

Have you found a solution yet?

[PanchoG17]

not yet, waiting for some advice from google team.
as soon i get a response i will share here my friend

[prashantkul]

Shouldn't the ADK/Agent be able to extract headers from the HTTP post request simply?

[seanzhou1023]

@PanchoG17 thank you for your question. the step 1, 2, 3 in your example flow need to be handled by your agent server (similar as ADK API server) which runs the agent (e.g. via ADK's Runner). And once user is verified you put a signal in the adk session state, and your tools that executes the CRUD operation should check this session state to see whether the signal is there , if it's not there, then your tool should reject the operation. Better put this signal prefixed with "temp:" in the session, if your agent session is not correlated with your user session, so that it won't be persisted.

[RubenAgDev]

That's precisely the question.
How should the Agent Server handle 1,2,3?

[MagnusDot]

Thanks for the clarification. Quick question, shouldn’t ADK handle HTTPBearer tokens natively?
Right now, the docs aren’t clear if the Authorization header is automatically propagated to tool_context. Having to manually validate and inject it into session state feels like something the framework could support by default.

Could this be made native or at least clarified in the docs?

[PanchoG17]

Hi! This is a simplified and untested extract from a working implementation.
It is shared for didactic purposes only, to illustrate how the pieces fit together rather than as production-ready code.

In this setup I integrate FastAPI with ADK agents, session persistence, and tool execution, including both MCPToolset-based tools and tools defined directly on the agent.

One of the main problems I needed to solve was how to propagate user context and authentication to tools. The approach that worked for me was to keep a login event stored in the session, updating the session state with user-related information. That state can then be accessed by tools defined on the agent itself. For MCP tools, authentication is configured separately through the toolset.

This is not meant to be the best or only solution — it’s simply the one that allowed the whole flow to work consistently in my case. The code has been reduced from a larger, working version, so there is definitely room for cleanup and improvement, but the core ideas and integration points are intentionally kept visible.

# main.py
import os
import time
import asyncio

from fastapi import FastAPI, Depends, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware

from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import DatabaseSessionService, Session
from google.adk.events import Event, EventActions
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
from google.adk.tools.openapi_tool.auth.auth_helpers import token_to_scheme_credential
from google.genai import types

from dependencies import get_user_jwt


# --------------------------
# Basic configuration
# --------------------------
APP_NAME = "basic_adk_example"
DB_STRING = os.getenv("DB_STRING")
MCP_HOST = os.getenv("MCP_HOST") #(fastMCP Server) 

# Lock global to avoid race conditions when mutating tool auth
_auth_lock = asyncio.Lock()


# --------------------------
# Agent definition
# --------------------------
agent = Agent(
    name="root_agent",
    model="gemini-2.0-flash",
    instruction="Answer to user and use provided tools",
    tools=[
        MCPToolset(
            connection_params=StreamableHTTPConnectionParams(
                url=f"{MCP_HOST}/mcp"
            )
        )
    ],
)


# --------------------------
# ADK services
# --------------------------
session_service = DatabaseSessionService(db_url=DB_STRING)

runner = Runner(
    agent=agent,
    app_name=APP_NAME,
    session_service=session_service,
)


# --------------------------
# FastAPI app
# --------------------------
app = FastAPI(title="ADK + FastAPI (basic example)")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_headers=["*"],
    allow_methods=["*"],
    allow_credentials=True,
)


# --------------------------
# Session + MCP auth handling
# --------------------------
async def config_session(
    session_id: str,
    user_id: int,
    auth_header: str,
) -> Session:
    """
    - Inject JWT into MCP tools
    - Load existing ADK session
    - Append a login_event (state_delta)
    """

    # Protect shared tool state
    async with _auth_lock:
        token = auth_header.split(" ")[1]

        # Build ADK auth objects from JWT
        auth_scheme, auth_credential = token_to_scheme_credential(
            "oauth2Token",
            "header",
            "Authorization",
            token,
        )

        # Apply auth to MCP tools
        for tool in agent.tools:
            if hasattr(tool, "_connection_params"):
                tool._auth_scheme = auth_scheme
                tool._auth_credential = auth_credential
                tool._connection_params.headers = {
                    "Authorization": f"Bearer {token}"
                }

    # Load session
    session = await session_service.get_session(
        app_name=APP_NAME,
        user_id=user_id,
        session_id=session_id,
    )

    if not session:
        raise HTTPException(400, "Invalid session_id")

    # --------------------------
    # Register login event
    # --------------------------
    """
    Note about login_event
    
    The login_event is still present, but it’s not strictly required. It was part of an earlier approach to propagate user context into the session, and I left it in place for now.
    
    With MCPToolset, token-based auth already covers most of this, so this event could likely be removed or simplified.
    """
    state_changes = {
        "task_status": "active",
        "user_id": user_id, #:temp key doesnt work for me
    }

    login_event = Event(
        invocation_id="login_event",
        author="root_agent",
        actions=EventActions(state_delta=state_changes),
        timestamp=time.time(),
    )

    await session_service.append_event(session, login_event)

    return session


# --------------------------
# Chat endpoint
# --------------------------
@app.post("/chat")
async def chat(request: Request, user_id: int = Depends(get_user_jwt)):
    body = await request.json()

    message = body["message"]
    session_id = body["session_id"]
    auth_header = request.headers.get("Authorization")

    # Configure session and MCP auth
    session = await config_session(session_id, user_id, auth_header)

    # Build user message
    content = types.Content(
        role="user",
        parts=[types.Part(text=message)],
    )

    # Run agent
    events = runner.run_async(
        user_id=user_id,
        session_id=session.id,
        new_message=content,
    )

    final_response = ""

    async for event in events:
        if event.is_final_response() and event.content:
            final_response = event.content.parts[0].text

    return {"response": final_response}