feat(a2a): Add A2A authentication middleware with TIP token propagation support

[loveyana]

Add A2A authentication middleware with TIP token propagation support

This PR introduces comprehensive A2A (Agent-to-Agent) authentication middleware with TIP (Trust Identity Propagation) token support for secure communication across the Volcengine Agent ecosystem.

Key Features

• A2AAuthMiddleware: Extracts auth tokens from requests and handles credential storage
• TIP Token Exchange: Automatic workload token generation via IdentityClient
• Multi-auth Support: Both Authorization header and query string authentication methods
• Credential Injection: Automatic credential injection into RemoteVeAgent HTTP clients
• Request Context: Sets user and auth information in request scope for downstream handlers

Usage Example

Basic A2A Server Setup with Authentication

from google.adk.a2a.utils.agent_to_a2a import to_a2a
from veadk.a2a.ve_middlewares import build_a2a_auth_middleware
from veadk.auth.credential_service import VeCredentialService
from veadk.runner import Runner
from veadk import Agent

# Initialize your agent
agent = Agent(name="my_secure_agent", tools=[...])

# Create credential service for managing authentication tokens
credential_service = VeCredentialService()

def create_runner() -> Runner:
    """Create a runner for the agent with credential service support."""
    return Runner(
        app_name=agent.name or "adk_agent",
        agent=agent,
        credential_service=credential_service,  # Enable credential management
    )

# Convert agent to A2A server
a2a_app = to_a2a(agent, runner=create_runner(), port=8001)

# Add authentication middleware for secure A2A communication
a2a_app.add_middleware(
    build_a2a_auth_middleware(
        app_name=agent.name,                    # App identifier for credential scoping
        credential_service=credential_service,  # Service for storing/retrieving credentials
        auth_method="header",                   # Support Authorization header (default)
        # auth_method="querystring",            # Alternative: query parameter auth
        # token_param="token",                  # Query param name (when using querystring)
    )
)

Client-side Remote Agent with Auto-authentication

from veadk.a2a.remote_ve_agent import RemoteVeAgent

# Remote agent automatically injects credentials from InvocationContext
remote_agent = RemoteVeAgent(
    name="secure_remote_agent",
    url="https://your-a2a-server.com",
    auth_method="header"  # Credentials injected at runtime from credential service
)

Technical Details

Authentication Flow

1. Request Processing: Middleware extracts JWT tokens from Authorization header or query parameters
2. User Identification: Parses JWT to extract user_id and delegation chain information
3. Credential Storage: Stores authentication credentials in VeCredentialService with app/user scoping
4. TIP Token Exchange: Exchanges TIP tokens for workload access tokens via IdentityClient

Credential Injection

• RemoteVeAgent automatically retrieves credentials from InvocationContext.credential_service
• Supports both header-based (Authorization: Bearer) and query parameter authentication
• TIP tokens are automatically injected via X-Ve-TIP-Token header for trust propagation

Security Features

• JWT validation and delegation chain extraction
• Scoped credential storage (app_name + user_id)
• Automatic workload token refresh via IdentityClient
• Secure credential propagation across A2A calls

Changes

• Add A2AAuthMiddleware class in veadk/a2a/ve_middlewares.py
• Add build_a2a_auth_middleware() factory function
• Enhance RemoteVeAgent with _inject_auth_token() method
• Add VeCredentialService integration with ADK BaseCredentialService
• Add comprehensive test coverage for authentication flows
• Add documentation and usage examples

This enables secure A2A communication with automatic credential propagation across the Volcengine Agent runtimes.

[cuericlee]

break change for existing a2a app.

[cuericlee]

import time
import requests
from google.adk.run import RemoteAgent

OAUTH_TOKEN_URL = "https://your-idp.com/oauth/token"
CLIENT_ID = "your-client-id"
CLIENT_SECRET = "your-client-secret"

class OAuthSession:
def init(self):
self.access_token = None
self.refresh_token = None
self.expires_at = 0

def authenticate(self):
    resp = requests.post(OAUTH_TOKEN_URL, data={
        "grant_type": "password",
        "username": "[email protected]",
        "password": "pass",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
    }).json()

    self.access_token = resp["access_token"]
    self.refresh_token = resp["refresh_token"]
    self.expires_at = time.time() + resp["expires_in"]

def refresh(self):
    resp = requests.post(OAUTH_TOKEN_URL, data={
        "grant_type": "refresh_token",
        "refresh_token": self.refresh_token,
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
    }).json()

    self.access_token = resp["access_token"]
    self.expires_at = time.time() + resp["expires_in"]

def get_token(self):
    if time.time() >= self.expires_at:
        self.refresh()
    return self.access_token

oauth = OAuthSession()
oauth.authenticate()

def build_agent():
return RemoteAgent(
url="https://remote-agent.company.com/mcp",
headers={
"Authorization": f"Bearer {oauth.get_token()}"
}
)

agent = build_agent()

[loveyana]

break change for existing a2a app.

good point, have fixed

[Copilot]

Pull Request Overview

This PR introduces comprehensive A2A (Agent-to-Agent) authentication middleware for the Volcengine Agent ecosystem, enabling secure service-to-service communication with TIP (Trust Identity Propagation) token support.

Key Changes:

• New authentication utilities for JWT parsing, delegation chain extraction, and flexible auth config building
• A2A authentication middleware that extracts credentials from requests and exchanges TIP tokens for workload access tokens
• VeCredentialService for app/user-scoped credential management extending ADK's BaseCredentialService
• Enhanced RemoteVeAgent with automatic credential injection from InvocationContext
• Global singleton IdentityClient pattern for improved connection pooling and credential caching

Reviewed Changes

Copilot reviewed 16 out of 16 changed files in this pull request and generated 10 comments.

Show a summary per file

File	Description
veadk/utils/auth.py	New authentication utilities module providing JWT parsing, delegation chain extraction, and AuthConfig building functions
veadk/tools/builtin_tools/agent_authorization.py	Refactored to use new auth utilities and support actor chain validation in permission checks
veadk/runner.py	Added credential_service parameter support for integration with VeCredentialService
veadk/integrations/ve_identity/token_manager.py	Updated type hints to support CallbackContext in addition to ToolContext and ReadonlyContext
veadk/integrations/ve_identity/identity_client.py	Improved error handling for STS operations and added original_callers parameter to check_permission
veadk/integrations/ve_identity/auth_processor.py	Refactored to use centralized get_default_identity_client helper
veadk/integrations/ve_identity/auth_mixins.py	Updated to use get_default_identity_client for consistent client management
veadk/integrations/ve_identity/auth_config.py	Added get_default_identity_client function to retrieve global IdentityClient instance
veadk/integrations/ve_identity/init.py	Exported get_default_identity_client utility for public use
veadk/configs/auth_configs.py	Added singleton IdentityClient management to VeIdentityConfig with lazy initialization
veadk/auth/credential_service.py	New VeCredentialService implementing app/user-scoped credential storage extending BaseCredentialService
veadk/a2a/ve_middlewares.py	New A2AAuthMiddleware for extracting auth tokens and exchanging TIP tokens with factory function
veadk/a2a/ve_a2a_server.py	Integrated credential_service parameter into VeA2AServer and init_app
veadk/a2a/remote_ve_agent.py	Enhanced with _inject_auth_token method for automatic credential injection from InvocationContext
tests/test_ve_a2a_middlewares.py	Comprehensive test suite for A2A authentication middleware
tests/auth/test_credential_service.py	Unit tests for VeCredentialService covering various credential management scenarios

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[cuericlee]

import time
import requests
from google.adk.run import RemoteAgent

OAUTH_TOKEN_URL = "https://your-idp.com/oauth/token"
CLIENT_ID = "your-client-id"
CLIENT_SECRET = "your-client-secret"

class OAuthSession:
def init(self):
self.access_token = None
self.refresh_token = None
self.expires_at = 0

def authenticate(self):
    resp = requests.post(OAUTH_TOKEN_URL, data={
        "grant_type": "password",
        "username": "[email protected]",
        "password": "pass",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
    }).json()

    self.access_token = resp["access_token"]
    self.refresh_token = resp["refresh_token"]
    self.expires_at = time.time() + resp["expires_in"]

def refresh(self):
    resp = requests.post(OAUTH_TOKEN_URL, data={
        "grant_type": "refresh_token",
        "refresh_token": self.refresh_token,
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
    }).json()

    self.access_token = resp["access_token"]
    self.expires_at = time.time() + resp["expires_in"]

def get_token(self):
    if time.time() >= self.expires_at:
        self.refresh()
    return self.access_token

oauth = OAuthSession()
oauth.authenticate()

def build_agent():
return RemoteAgent(
url="https://remote-agent.company.com/mcp",
headers={
"Authorization": f"Bearer {oauth.get_token()}"
}
)

agent = build_agent()

[cuericlee]

don't inject the RemoteA2aAgent, with fetch_token() instead.
remote = RemoteA2aAgent(
name="secure-agent",
url="https://my-secure-agent.example.com/a2a",
headers={"Authorization": f"Bearer {fetch_token()}"},
)

[cuericlee]

only support header with Authorization, for querystring using url only

[loveyana]

don't inject the RemoteA2aAgent, with fetch_token() instead. remote = RemoteA2aAgent( name="secure-agent", url="https://my-secure-agent.example.com/a2a", headers={"Authorization": f"Bearer {fetch_token()}"}, )

@cuericlee 在 RemoteA2aAgent 初始化时，fetch_token() 只会被调用一次，返回的 token 会被固定在 headers 中。之后即使 token 过期或需要刷新，headers 也不会更新。不能这么搞的

[cuericlee]

why not extend from InMemoryCredentialService
https://github.com/google/adk-python/blob/main/src/google/adk/auth/credential_service/in_memory_credential_service.py#L29

[loveyana]

@cuericlee since InMemoryCredentialService is an experimental feature, and only the abstract class BaseCredentialService needs to be passed at all locations. to prevent the impact caused by changes to InMemoryCredentialService, VeCredentialService is separately defined based on the abstract definition.

[cuericlee]

/lgtm

[cuericlee]

/lgtm

[cuericlee]

/lgtm