How to pass a bearer token to tools when a bearer token is specified in Authorization HTTP Header in a request from client to a adk application?

[kyu08]

Thank you for maintaining ADK always!

Situation

We are building an application like below. ADK application receives request from frontend and it calls backend API within tool.

We pass a bearer token which specified as Authorization HTTP Header when we request from frontend to ADK application.
And then we want to specify Authorization HTTP Header when we request to the backend API within ADK tool functions.

Question

I didn't understand how to pass a bearer token which came from frontend to a tool which calls backend API using adk-python though I read and search GitHub issue and PRs.
Or could we pass a bearer token as a part of request body to ADK? Or could we make a HTTP request to ADK to save a token to memory before invoking agent?
Could you please tell me how?

Thank you.

[adk-bot]

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

TLDR: Use ADK's Session State to pass the bearer token from your web server endpoint to your tool. Your server code will add the token to the state, and your tool will read it from the ToolContext.

---

Hello! Thank you for the great question. This is a common and important pattern for building secure applications with ADK. Here’s a detailed guide on how you can achieve this.

The recommended approach is to use the Session State to pass contextual data, like an authentication token, from your application's entrypoint (e.g., your FastAPI endpoint) down to your tool functions.

Step 1: Capture the Token in Your Application Endpoint

First, in your web server code, you need to capture the bearer token from the Authorization header. You then add this token to the session state before running the agent.

Here is a conceptual example using FastAPI:

from fastapi import FastAPI, Header, Depends
from typing import Annotated
from google.adk.runners import Runner
from google.adk.sessions import Session

# Assume 'runner' and 'session_service' are already configured
# runner: Runner = ...
# session_service: SessionService = ...

app = FastAPI()

@app.post("/invoke-agent")
async def invoke_agent(
    # Get the current session (you might have your own dependency for this)
    session: Annotated[Session, Depends(get_or_create_session)],
    # Extract the Authorization header
    authorization: Annotated[str | None, Header()] = None
):
    # 1. Add the token to the session state
    if authorization:
        session.state["bearer_token"] = authorization

    # The session with the updated state will be used by the runner
    # 2. Now, run the agent as usual
    # The runner will automatically use the updated session object
    async for event in runner.run_async(
        session=session,
        # ... other params
    ):
        # process events
        pass

    return {"status": "ok"}

Step 2: Access the Token in Your Tool

Next, modify your tool's function signature to accept tool_context: ToolContext. ADK will automatically inject this object at runtime, giving you access to the session state.

import requests
from google.adk.tools import ToolContext, tool

@tool
def call_backend_api(query: str, tool_context: ToolContext) -> dict:
    '''Calls the secure backend API.'''

    # 1. Retrieve the token from the state
    auth_header = tool_context.state.get("bearer_token")

    if not auth_header:
        return {"error": "Authentication token not found in state."}

    headers = {
        "Authorization": auth_header,
        "Content-Type": "application/json"
    }
    
    try:
        # 2. Use the token to make the authenticated request
        response = requests.post(
            "https://your-backend-api.com/endpoint",
            headers=headers,
            json={"query": query}
        )
        response.raise_for_status()  # Raise an exception for bad status codes
        return response.json()

    except requests.exceptions.RequestException as e:
        return {"error": f"API call failed: {e}"}

Why This Approach Works Best

• Decoupling: Your tool's logic remains independent of the web framework. It only needs to know about the ToolContext, not about HTTP headers or FastAPI. [1, 4]
• State Management: This leverages ADK's built-in session management, which is the intended way to handle conversational memory and context. [1, 8]
• Security: You are explicitly passing the credential for the current session, which is a secure and clear pattern.

For more details, you can refer to the official documentation on Context and Tools.

I hope this helps you build your application!

[1] https://google.github.io/adk-docs/context/
[4] https://google.github.io/adk-docs/tools/
[8] https://google.github.io/adk-docs/get-started/about/

[RubenAgDev]

This requires to re-do every single endpoint in the FastAPI app that the ADK provides

[mkkhedawat]

Does LLM have access to ToolContext in above pattern ?

[achandrase-chwy]

Hi @kyu08, @RubenAgDev,

You can use FastAPI middleware to load the bearer token into contextvars (request scope). This allows you to make the token available across your request lifecycle and the code to read and hold token will be in just one place and not in each route. Then, you can use the before_tool_callback hook to fetch the token from the context and pass it into your tool calls.

Keep in mind that the bearer token may expire between when you receive it and when it’s used within a tool call — especially if the call involves a downstream API request.

A more robust design is to have your agent or a gateway in front of your agent handle the incoming bearer token. That component should:
• Validate the token and extract any custom claims.
• Use a OAuth 2.0 machine-to-machine (M2M) or client credentials flow for any downstream tool or API calls.

This avoids having to propagate short-lived user tokens through the agent, reducing both complexity and potential security issues.

[diabllos]

I've been looking into a similar thing. This is what I've came up with., It's not production ready and for sure could be done way better, but it seems to work. Most of the code has been borrowed from:
2. Handling the Interactive OAuth/OIDC Flow (Client-Side)

from google.adk.tools.openapi_tool.auth.auth_helpers import token_to_scheme_credential

adk_events = await self._send_to_adk_agent(adk_session_id, user_id, content)
                
for event in adk_events:
	# Use helper to check for the specific auth request event
	if (auth_request_function_call := self._get_auth_request_function_call(event)):
		print("--> Authentication required by agent.")
		auth_request_function_call_id = auth_request_function_call['id']
		auth_conf = self._get_auth_config(auth_request_function_call)
		auth_schema, auth_creds = token_to_scheme_credential(token_type="oauth2Token", credential_value=authorization)
		
		auth_conf.auth_scheme = auth_schema
		auth_conf.exchanged_auth_credential = auth_creds

		auth_content = types.Content(
			role='user', # Role can be 'user' when sending a FunctionResponse
			parts=[
				types.Part(
					function_response=types.FunctionResponse(
						id=auth_request_function_call_id,       # Link to the original request
						name='adk_request_credential', # Special framework function name
						response=auth_conf.model_dump() # Send back the *updated* AuthConfig
					)
				)
			],
		).model_dump_json()
		adk_events = await self._send_to_adk_agent(adk_session_id, user_id, json.loads(auth_content))

Functions _get_auth_config _get_auth_request_function_call are within the helpers.py in the previously mentioned docs.
_send_to_adk_agent is just a wrapper on httpx call to /run

async def _send_to_adk_agent(self, adk_session_id: str, user_id: str, content: Dict[str, Any], authorization: Optional[str] = None) -> List[Dict]:
    """Send message to ADK-Agent and get response"""
    try:
        # Prepare ADK RunAgentRequest
        run_request = {
            "appName": self.adk_app_name,
            "userId": user_id,
            "sessionId": adk_session_id,
            "newMessage": content,
            "streaming": False
        }
        
        # Prepare headers for ADK-Agent request
        headers = {}
        if authorization:
            headers["Authorization"] = authorization
            logger.info(f"Forwarding authorization token to ADK-Agent for message processing")
        
        # Log the request being sent to ADK-Agent
        logger.info(f"=== Sending to ADK-Agent ===")
        logger.info(f"URL: {self.adk_agent_base_url}/run")
        logger.info(f"Payload: {json.dumps(run_request, indent=2)}")
        if authorization:
            logger.info(f"Authorization: {authorization[:20]}...")  # Log first 20 chars for security
        logger.info(f"===========================")
        
        # Call ADK /run endpoint
        run_url = f"{self.adk_agent_base_url}/run"
        response = await self.http_client.post(run_url, json=run_request, headers=headers)
        
        if response.status_code == 200:
            events = response.json()
            logger.info(f"ADK response received with {len(events)} events")
            return events
        else:
            logger.error(f"ADK-Agent returned error: {response.status_code} - {response.text}")
            raise HTTPException(status_code=response.status_code, detail="ADK-Agent processing failed")
            
    except httpx.RequestError as e:
        logger.error(f"Failed to communicate with ADK-Agent: {e}")
        raise HTTPException(status_code=503, detail="ADK-Agent unavailable")

The only thing that really needs to be fixed (besides the code above ;)) - is that when a request is authenticated and passed to the tool - in our case an MCP tool, it has the Authorization header with 2 Bearer words in it, no idea why... yet (it could be caused smtg else we do in the code). Just FYI.