Pass external API tokens to MCP tools via X-External-Token header using ADK session

adk/README.md

@@ -9,6 +9,7 @@ SDK for Google ADK that helps to get agents configured in the Agentic Layer quic
 - Configures A2A protocol for inter-agent communication
 - Offers parsing methods for sub agents and tools
 - Set log level via env var `LOGLEVEL` (default: `INFO`)
+- Automatically passes external API tokens to MCP tools via the `X-External-Token` header
 
 ## Usage
 
@@ -100,3 +101,53 @@ Body logging behavior:
 
 **Note**: Starlette body logging is more limited than HTTPX because it must avoid consuming request/response streams.
 Bodies are only captured when already buffered in the ASGI scope.
+
+## External API Token Passing
+
+The SDK supports passing external API tokens from A2A requests to MCP tools. This enables MCP servers to authenticate with external APIs on behalf of users.
+
+### How It Works
+
+1. **Token Capture**: When an A2A request includes the `X-External-Token` header, the SDK automatically captures and stores it in the ADK session state
+2. **Secure Storage**: The token is stored in ADK's session management system with a private key prefix (`__external_token__`), making it inaccessible to agent code
+3. **Automatic Injection**: When MCP tools are invoked, the SDK uses ADK's `header_provider` hook to retrieve the token from the session and inject it as the `X-External-Token` header in tool requests
+4. **Propagation**: The token is passed to all connected MCP servers and sub-agents for the duration of the session
+
+### Usage Example
+
+Simply include the `X-External-Token` header in your A2A requests:
+
+```bash
+curl -X POST http://localhost:8000/ \
+  -H "Content-Type: application/json" \
+  -H "X-External-Token: your-api-token-here" \
+  -d '{
+    "jsonrpc": "2.0",
+    "id": 1,
+    "method": "message/send",
+    "params": {
+      "message": {
+        "role": "user",
+        "parts": [{"kind": "text", "text": "Your message"}],
+        "messageId": "msg-123",
+        "contextId": "ctx-123"
+      }
+    }
+  }'
+```
+
+The SDK will automatically pass `your-api-token-here` to all MCP tool calls made during that session.
+
+### Implementation Details
+
+- **TokenCapturingA2aAgentExecutor**: Custom A2A executor that extends `A2aAgentExecutor` to intercept requests and store the token in the ADK session state
+- **Session Storage**: Token stored in `session.state["__external_token__"]` using ADK's built-in session management
+- **Header Provider**: MCP tools configured with a `header_provider` function that reads from the session state via `ReadonlyContext`
+- **ADK Integration**: Uses ADK's official hooks and session mechanisms rather than external context variables
+
+### Security Considerations
+
+- Tokens are stored in ADK session state with a private key prefix (`__external_token__`)
+- Tokens are not directly accessible to agent code through normal session state queries
+- Tokens persist for the session duration and are managed by ADK's session lifecycle
+- This is a simple authentication mechanism; for production use, consider implementing more sophisticated authentication and authorization schemes

adk/agenticlayer/agent.py

@@ -9,17 +9,42 @@
 from a2a.utils.constants import AGENT_CARD_WELL_KNOWN_PATH
 from google.adk.agents import BaseAgent, LlmAgent
 from google.adk.agents.llm_agent import ToolUnion
+from google.adk.agents.readonly_context import ReadonlyContext
 from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
 from google.adk.tools.agent_tool import AgentTool
 from google.adk.tools.mcp_tool import StreamableHTTPConnectionParams
 from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
 from httpx_retries import Retry, RetryTransport
 
 from agenticlayer.config import InteractionType, McpTool, SubAgent
+from agenticlayer.constants import EXTERNAL_TOKEN_SESSION_KEY
 
 logger = logging.getLogger(__name__)
 
 
+def _get_mcp_headers_from_session(readonly_context: ReadonlyContext) -> dict[str, str]:
+    """Header provider function for MCP tools that retrieves token from ADK session.
+
+    This function is called by the ADK when MCP tools are invoked. It reads the
+    external token from the session state where it was stored during request
+    processing by TokenCapturingA2aAgentExecutor.
+
+    Args:
+        readonly_context: The ADK ReadonlyContext providing access to the session
+
+    Returns:
+        A dictionary of headers to include in MCP tool requests.
+        If a token is stored in the session, includes it in the headers.
+    """
+    # Access the session state through the readonly context
+    # The session state is a dict that can contain the external token
+    if readonly_context and readonly_context.session:
+        external_token = readonly_context.session.state.get(EXTERNAL_TOKEN_SESSION_KEY)
+        if external_token:
+            return {"X-External-Token": external_token}
+    return {}
+
+
 class AgentFactory:
     def __init__(
         self,
@@ -110,6 +135,8 @@ def load_tools(self, mcp_tools: list[McpTool]) -> list[ToolUnion]:
                         url=str(tool.url),
                         timeout=tool.timeout,
                     ),
+                    # Provide header provider to inject session-stored token into tool requests
+                    header_provider=_get_mcp_headers_from_session,
                 )
             )
 

adk/agenticlayer/agent_to_a2a.py

@@ -7,11 +7,13 @@
 import logging
 from typing import AsyncIterator, Awaitable, Callable
 
+from a2a.server.agent_execution.context import RequestContext
 from a2a.server.apps import A2AStarletteApplication
 from a2a.server.request_handlers import DefaultRequestHandler
 from a2a.server.tasks import InMemoryTaskStore
 from a2a.types import AgentCapabilities, AgentCard
 from a2a.utils.constants import AGENT_CARD_WELL_KNOWN_PATH
+from google.adk.a2a.converters.request_converter import AgentRunRequest
 from google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor
 from google.adk.agents import LlmAgent
 from google.adk.agents.base_agent import BaseAgent
@@ -21,15 +23,63 @@
 from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
 from google.adk.runners import Runner
 from google.adk.sessions.in_memory_session_service import InMemorySessionService
+from google.adk.sessions.session import Session
 from starlette.applications import Starlette
 
 from .agent import AgentFactory
 from .callback_tracer_plugin import CallbackTracerPlugin
 from .config import McpTool, SubAgent
+from .constants import EXTERNAL_TOKEN_SESSION_KEY
 
 logger = logging.getLogger(__name__)
 
 
+class TokenCapturingA2aAgentExecutor(A2aAgentExecutor):
+    """Custom A2A agent executor that captures and stores the X-External-Token header.
+
+    This executor extends the standard A2aAgentExecutor to intercept the request
+    and store the X-External-Token header in the ADK session state. This allows
+    MCP tools to access the token via the header_provider hook, using ADK's
+    built-in session management rather than external context variables.
+    """
+
+    async def _prepare_session(
+        self,
+        context: RequestContext,
+        run_request: AgentRunRequest,
+        runner: Runner,
+    ) -> Session:
+        """Prepare the session and store the external token if present.
+
+        This method extends the parent implementation to capture the X-External-Token
+        header from the request context and store it in the session state.
+
+        Args:
+            context: The A2A request context containing the call context with headers
+            run_request: The agent run request
+            runner: The ADK runner instance
+
+        Returns:
+            The prepared session with the external token stored in its state
+        """
+        # Call parent to get or create the session
+        session: Session = await super()._prepare_session(context, run_request, runner)
+
+        # Extract the X-External-Token header from the request context
+        # The call_context.state contains headers from the original HTTP request
+        if context.call_context and "headers" in context.call_context.state:
+            headers = context.call_context.state["headers"]
+            external_token = headers.get("x-external-token")  # HTTP headers are case-insensitive
+
+            if external_token:
+                # Store the token in the session state with a private key
+                # The session state is mutable and changes are persisted automatically
+                session.state[EXTERNAL_TOKEN_SESSION_KEY] = external_token
+                logger.debug("Stored external token in session %s", session.id)
+
+        return session
+
+
 class HealthCheckFilter(logging.Filter):
     def filter(self, record: logging.LogRecord) -> bool:
         # Check if the log message contains the well known path of the card, which is used for health checks
@@ -55,15 +105,16 @@ async def create_runner() -> Runner:
                 plugins=[CallbackTracerPlugin()],
             ),
             artifact_service=InMemoryArtifactService(),
-            session_service=InMemorySessionService(),  # type: ignore
-            memory_service=InMemoryMemoryService(),  # type: ignore
-            credential_service=InMemoryCredentialService(),  # type: ignore
+            session_service=InMemorySessionService(),  # type: ignore[no-untyped-call]
+            memory_service=InMemoryMemoryService(),  # type: ignore[no-untyped-call]
+            credential_service=InMemoryCredentialService(),  # type: ignore[no-untyped-call]
         )
 
     # Create A2A components
     task_store = InMemoryTaskStore()
 
-    agent_executor = A2aAgentExecutor(
+    # Use custom executor that captures X-External-Token and stores in session
+    agent_executor = TokenCapturingA2aAgentExecutor(
         runner=create_runner,
     )
 

adk/agenticlayer/constants.py

@@ -0,0 +1,4 @@
+"""Constants shared across the agenticlayer package."""
+
+# Key used to store the external token in the ADK session state
+EXTERNAL_TOKEN_SESSION_KEY = "__external_token__"  # nosec B105

adk/tests/test_external_token.py

@@ -0,0 +1,68 @@
+"""Tests for external token passing to MCP tools via ADK session."""
+
+from agenticlayer.agent import _get_mcp_headers_from_session
+from agenticlayer.constants import EXTERNAL_TOKEN_SESSION_KEY
+from google.adk.sessions.session import Session
+
+
+def test_header_provider_retrieves_token_from_session() -> None:
+    """Test that the header provider function can retrieve token from session state."""
+    # Given: A session with an external token stored
+    test_token = "test-api-token-xyz"  # nosec B105
+    session = Session(
+        id="test-session",
+        app_name="test-app",
+        user_id="test-user",
+        state={EXTERNAL_TOKEN_SESSION_KEY: test_token},
+        events=[],
+        last_update_time=0.0,
+    )
+
+    # Create a mock readonly context
+    class MockReadonlyContext:
+        def __init__(self, session: Session) -> None:
+            self.session = session
+
+    readonly_context = MockReadonlyContext(session)
+
+    # When: Calling the header provider function
+    headers = _get_mcp_headers_from_session(readonly_context)  # type: ignore[arg-type]
+
+    # Then: The headers should include the X-External-Token
+    assert headers == {"X-External-Token": test_token}
+
+
+def test_header_provider_returns_empty_when_no_token() -> None:
+    """Test that the header provider returns empty dict when no token is present."""
+    # Given: A session without an external token
+    session = Session(
+        id="test-session",
+        app_name="test-app",
+        user_id="test-user",
+        state={},  # No token
+        events=[],
+        last_update_time=0.0,
+    )
+
+    # Create a mock readonly context
+    class MockReadonlyContext:
+        def __init__(self, session: Session) -> None:
+            self.session = session
+
+    readonly_context = MockReadonlyContext(session)
+
+    # When: Calling the header provider function
+    headers = _get_mcp_headers_from_session(readonly_context)  # type: ignore[arg-type]
+
+    # Then: The headers should be empty
+    assert headers == {}
+
+
+def test_header_provider_handles_none_context() -> None:
+    """Test that the header provider safely handles None context."""
+    # When: Calling the header provider with None
+    headers = _get_mcp_headers_from_session(None)  # type: ignore[arg-type]
+
+    # Then: The headers should be empty (no exception)
+    assert headers == {}
+