feat(mcp): add credential_key shorthand for Bearer token propagation

Adds `credential_key` parameter to McpToolset and McpToolsetConfig as
a convenience shorthand that reads a token from session state and sends
it as an `Authorization: Bearer <token>` header. Internally creates a
header_provider, so it combines cleanly with existing header_provider
and state_header_mapping options. Aligns with credential_key usage on
other ADK toolsets (e.g., OpenAPIToolset). Addresses #5103.

Author: timof1308

src/google/adk/tools/mcp_tool/mcp_toolset.py

@@ -253,6 +253,7 @@ def __init__(
       auth_credential: Optional[AuthCredential] = None,
       require_confirmation: Union[bool, Callable[..., bool]] = False,
       header_provider: Optional[HeaderProvider] = None,
+      credential_key: Optional[str] = None,
       progress_callback: Optional[
           Union[ProgressFnT, ProgressCallbackFactory]
       ] = None,
@@ -283,6 +284,11 @@ def __init__(
         Can be a single boolean or a callable to apply to all tools.
       header_provider: A callable that takes a ReadonlyContext and returns a
         dictionary of headers to be used for the MCP session.
+      credential_key: A session state key whose value is sent as an
+        ``Authorization: Bearer <token>`` header on every MCP request. This is
+        a convenience shorthand that internally creates a header_provider. If
+        both ``credential_key`` and ``header_provider`` are provided, they are
+        combined.
       progress_callback: Optional callback to receive progress notifications
         from MCP server during long-running tool execution. Can be either:  - A
         ``ProgressFnT`` callback that receives (progress, total, message). This
@@ -310,7 +316,20 @@ def __init__(
 
     self._connection_params = connection_params
     self._errlog = errlog
-    self._header_provider = header_provider
+
+    # Build the effective header_provider from credential_key and/or the
+    # explicit header_provider parameter.
+    credential_provider = (
+        create_session_state_header_provider(state_key=credential_key)
+        if credential_key
+        else None
+    )
+    if credential_provider and header_provider:
+      self._header_provider = create_combined_header_provider(
+          [credential_provider, header_provider]
+      )
+    else:
+      self._header_provider = credential_provider or header_provider
     self._progress_callback = progress_callback
 
     # Create the session manager that will handle the MCP connection
@@ -580,13 +599,14 @@ def from_config(
     else:
       raise ValueError("No connection params found in McpToolsetConfig.")
 
-    # Create header_provider from state_header_mapping if specified
-    header_provider = None
+    # Build header_provider from state_header_mapping and/or credential_key.
+    providers = []
+
     if mcp_toolset_config.state_header_mapping:
       state_mapping = mcp_toolset_config.state_header_mapping
       state_format = mcp_toolset_config.state_header_format or {}
 
-      providers = [
+      providers.extend([
           create_session_state_header_provider(
               state_key=state_key,
               header_name=header_name,
@@ -595,9 +615,18 @@ def from_config(
               strict=mcp_toolset_config.state_header_strict,
           )
           for state_key, header_name in state_mapping.items()
-      ]
+      ])
 
-      header_provider = create_combined_header_provider(providers)
+    if mcp_toolset_config.credential_key:
+      providers.append(
+          create_session_state_header_provider(
+              state_key=mcp_toolset_config.credential_key,
+          )
+      )
+
+    header_provider = (
+        create_combined_header_provider(providers) if providers else None
+    )
 
     return cls(
         connection_params=connection_params,
@@ -645,6 +674,12 @@ class McpToolsetConfig(BaseToolConfig):
 
   use_mcp_resources: bool = False
 
+  credential_key: Optional[str] = None
+  """A session state key whose value is sent as an ``Authorization: Bearer``
+  header on every MCP HTTP request. Convenience shorthand that is equivalent
+  to ``state_header_mapping: {<key>: Authorization}`` with
+  ``state_header_format: {Authorization: "Bearer {value}"}``."""
+
   state_header_mapping: Optional[Dict[str, str]] = None
   """Maps session state keys to HTTP header names.
 

tests/unittests/tools/mcp_tool/test_jwt_token_propagation.py

@@ -429,6 +429,53 @@ def test_from_config_no_state_mapping_no_provider(self):
     # No header provider should be created
     assert toolset._header_provider is None
 
+  def test_from_config_with_credential_key(self):
+    """Test that from_config creates header provider from credential_key."""
+    from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
+    from google.adk.tools.tool_configs import ToolArgsConfig
+
+    config = ToolArgsConfig(
+        stdio_server_params={"command": "test_command", "args": []},
+        credential_key="my_token",
+    )
+
+    toolset = McpToolset.from_config(config, "/fake/path")
+
+    assert toolset._header_provider is not None
+
+    mock_context = Mock(spec=ReadonlyContext)
+    mock_context.state = {"my_token": "test-jwt-123"}
+
+    headers = toolset._header_provider(mock_context)
+
+    assert headers == {"Authorization": "Bearer test-jwt-123"}
+
+  def test_from_config_credential_key_with_state_header_mapping(self):
+    """Test that credential_key and state_header_mapping combine."""
+    from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
+    from google.adk.tools.tool_configs import ToolArgsConfig
+
+    config = ToolArgsConfig(
+        stdio_server_params={"command": "test_command", "args": []},
+        credential_key="jwt_token",
+        state_header_mapping={"tenant_id": "X-Tenant-ID"},
+    )
+
+    toolset = McpToolset.from_config(config, "/fake/path")
+
+    mock_context = Mock(spec=ReadonlyContext)
+    mock_context.state = {
+        "jwt_token": "my-jwt",
+        "tenant_id": "tenant-42",
+    }
+
+    headers = toolset._header_provider(mock_context)
+
+    assert headers == {
+        "Authorization": "Bearer my-jwt",
+        "X-Tenant-ID": "tenant-42",
+    }
+
   def test_from_config_with_strict_mode(self):
     """Test that from_config respects state_header_strict setting."""
     from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
@@ -453,6 +500,77 @@ def test_from_config_with_strict_mode(self):
     assert "dict" in str(exc_info.value)
 
 
+class TestCredentialKey:
+  """Test suite for credential_key on McpToolset.__init__."""
+
+  def test_credential_key_creates_bearer_header(self):
+    """Test credential_key reads token from state and sends as Bearer."""
+    from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
+
+    toolset = McpToolset(
+        connection_params=StdioServerParameters(command="echo", args=[]),
+        credential_key="auth_token",
+    )
+
+    assert toolset._header_provider is not None
+
+    mock_context = Mock(spec=ReadonlyContext)
+    mock_context.state = {"auth_token": "my-jwt-token"}
+
+    headers = toolset._header_provider(mock_context)
+
+    assert headers == {"Authorization": "Bearer my-jwt-token"}
+
+  def test_credential_key_missing_state_returns_empty(self):
+    """Test credential_key returns empty headers when key not in state."""
+    from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
+
+    toolset = McpToolset(
+        connection_params=StdioServerParameters(command="echo", args=[]),
+        credential_key="auth_token",
+    )
+
+    mock_context = Mock(spec=ReadonlyContext)
+    mock_context.state = {}
+
+    headers = toolset._header_provider(mock_context)
+
+    assert headers == {}
+
+  def test_credential_key_none_means_no_provider(self):
+    """Test that credential_key=None does not create a provider."""
+    from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
+
+    toolset = McpToolset(
+        connection_params=StdioServerParameters(command="echo", args=[]),
+        credential_key=None,
+    )
+
+    assert toolset._header_provider is None
+
+  def test_credential_key_combines_with_header_provider(self):
+    """Test that credential_key and header_provider are combined."""
+    from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
+
+    custom_provider = lambda ctx: {"X-Custom": "value"}
+
+    toolset = McpToolset(
+        connection_params=StdioServerParameters(command="echo", args=[]),
+        credential_key="auth_token",
+        header_provider=custom_provider,
+    )
+
+    mock_context = Mock(spec=ReadonlyContext)
+    mock_context.state = {"auth_token": "my-jwt"}
+
+    headers = toolset._header_provider(mock_context)
+
+    assert headers == {
+        "Authorization": "Bearer my-jwt",
+        "X-Custom": "value",
+    }
+
+
 class TestRFC7230Compliance:
   """Test suite for RFC 7230 compliant header handling."""