♻️ Feat: support a2a custom authorization

backend/apps/a2a_client_app.py

@@ -6,7 +6,7 @@
 """
 import logging
 import uuid
-from typing import Annotated, List, Optional
+from typing import Annotated, Dict, List, Optional
 from http import HTTPStatus
 
 from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request
@@ -27,9 +27,10 @@
 
 
 class DiscoverFromUrlRequest(BaseModel):
-    """Request to discover external A2A agent from URL."""
+    """Request to discover an external A2A agent from an Agent Card URL."""
     url: str
     name: Optional[str] = None
+    custom_headers: Optional[Dict[str, str]] = None
 
 
 class DiscoverFromNacosRequest(BaseModel):
@@ -46,6 +47,11 @@
     )
 
 
+class UpdateExternalAgentSecurityCredentialsRequest(BaseModel):
+    """Request to configure credentials required by an Agent Card."""
+    security_credentials: Dict[str, str] = Field(default_factory=dict)
+
+
 class TestNacosConnectionRequest(BaseModel):
     """Request to test Nacos connectivity without saving the config."""
     nacos_addr: str = Field(description="Nacos server address (e.g., http://nacos-server:8848)")
@@ -74,7 +80,8 @@
         result = await a2a_client_service.discover_from_url(
             url=request.url,
             tenant_id=tenant_id,
-            user_id=user_id
+            user_id=user_id,
+            custom_headers=request.custom_headers
         )
 
         return JSONResponse(
@@ -203,6 +210,69 @@
         )
 
 
+@router.put("/agents/{external_agent_id}/security-credentials")
+async def update_external_agent_security_credentials(
+    external_agent_id: int,
+    request: UpdateExternalAgentSecurityCredentialsRequest,
+    authorization: Annotated[Optional[str], Header()] = None,
+    http_request: Request = None,
+):
+    """Save the credential values required to call an external A2A agent."""
+    try:
+        user_id, tenant_id, _ = get_current_user_info(authorization, http_request)
+        if not request.security_credentials:
+            raise HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST,
+                detail="At least one security credential is required"
+            )
+        if any(not scheme_id or not value for scheme_id, value in request.security_credentials.items()):
+            raise HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST,
+                detail="Security credential names and values must be non-empty"
+            )
+        agent = a2a_agent_db.get_external_agent_by_id(external_agent_id, tenant_id)
+        if not agent:
+            raise HTTPException(
+                status_code=HTTPStatus.NOT_FOUND,
+                detail=f"Agent {external_agent_id} not found"
+            )
+        supported_scheme_ids = set((agent.get("security_schemes") or {}).keys())
+        if not set(request.security_credentials).issubset(supported_scheme_ids):
+            raise HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST,
+                detail="Security credentials must use scheme names declared by the Agent Card"
+            )
+        result = a2a_agent_db.update_external_agent_security_credentials(
+            external_agent_id=external_agent_id,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            security_credentials=request.security_credentials,
+        )
+        if not result:
+            raise HTTPException(
+                status_code=HTTPStatus.NOT_FOUND,
+                detail=f"Agent {external_agent_id} not found"
+            )
+        agent = a2a_agent_db.get_external_agent_by_id(external_agent_id, tenant_id)
+        return JSONResponse(
+            status_code=HTTPStatus.OK,
+            content={
+                "status": "success",
+                "data": {
+                    "configured_security_scheme_ids": agent["configured_security_scheme_ids"],
+                },
+            },
+        )
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Update agent security credentials failed: {e}", exc_info=True)
+        raise HTTPException(
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
+            detail="Failed to update agent security credentials"
+        )
+
+
 @router.post("/agents/{external_agent_id}/refresh")
 async def refresh_agent_card(
     external_agent_id: int,

backend/consts/a2a_models.py

@@ -133,9 +133,13 @@ class A2AAgentCard(BaseModel):
 # =============================================================================
 
 class DiscoverFromUrlRequest(BaseModel):
-    """Request to discover an external A2A agent from URL."""
+    """Request to discover an external A2A agent from an Agent Card URL."""
     url: str = Field(description="Direct URL to the Agent Card")
     name: Optional[str] = Field(default=None, description="Optional display name override")
+    custom_headers: Optional[Dict[str, str]] = Field(
+        default=None,
+        description="Headers saved for Agent Card discovery and refresh only"
+    )
 
 
 class DiscoverFromNacosRequest(BaseModel):

backend/database/a2a_agent_db.py

@@ -145,18 +145,26 @@
     return PROTOCOL_JSONRPC
 
 
+def _configured_security_scheme_ids(security_credentials: Optional[Dict[str, str]]) -> List[str]:
+    """Return credential scheme IDs without exposing their secret values."""
+    return sorted(scheme_id for scheme_id, value in (security_credentials or {}).items() if value)
+
+
 def create_external_agent_from_url(
     source_url: str,
     name: str,
     description: Optional[str],
     agent_url: str,
     tenant_id: str,
     user_id: str,
     raw_card: Optional[Dict[str, Any]] = None,
     version: Optional[str] = None,
     streaming: bool = False,
     supported_interfaces: Optional[List[Dict[str, Any]]] = None,
     base_url: Optional[str] = None,
+    agent_card_headers: Optional[Dict[str, str]] = None,
+    security_schemes: Optional[Dict[str, Any]] = None,
+    security_requirements: Optional[List[Dict[str, Any]]] = None,
 ) -> Dict[str, Any]:
     """Create or update an external A2A agent discovered from URL.
 
@@ -172,6 +180,9 @@
         streaming: Whether this agent supports SSE streaming.
         supported_interfaces: All supported protocol interfaces.
         base_url: Base URL for health checks (service root address).
+        agent_card_headers: Headers saved only for Agent Card discovery and refresh.
+        security_schemes: Security schemes declared by the Agent Card.
+        security_requirements: Security requirements declared by the Agent Card.
 
     Returns:
         Created agent information dict.
@@ -202,6 +213,10 @@
             existing.streaming = streaming
             existing.supported_interfaces = supported_interfaces
             existing.raw_card = raw_card
+            existing.security_schemes = security_schemes
+            existing.security_requirements = security_requirements
+            if agent_card_headers is not None:
+                existing.agent_card_headers = agent_card_headers
             existing.cached_at = now
             existing.cache_expires_at = expires_at
             existing.updated_by = user_id
@@ -220,6 +235,9 @@
                 supported_interfaces=supported_interfaces,
                 source_type="url",
                 source_url=source_url,
+                agent_card_headers=agent_card_headers,
+                security_schemes=security_schemes,
+                security_requirements=security_requirements,
                 tenant_id=tenant_id,
                 created_by=user_id,
                 updated_by=user_id,
@@ -242,6 +260,9 @@
             "protocol_type": agent.protocol_type,
             "streaming": agent.streaming,
             "supported_interfaces": agent.supported_interfaces,
+            "security_schemes": agent.security_schemes,
+            "security_requirements": agent.security_requirements,
+            "configured_security_scheme_ids": _configured_security_scheme_ids(agent.security_credentials),
             "source_type": agent.source_type,
             "base_url": agent.base_url,
             "is_available": agent.is_available,
@@ -251,18 +272,20 @@
 
 
 def create_external_agent_from_nacos(
     name: str,
     description: Optional[str],
     agent_url: str,
     nacos_config_id: str,
     nacos_agent_name: str,
     tenant_id: str,
     user_id: str,
     raw_card: Optional[Dict[str, Any]] = None,
     version: Optional[str] = None,
     streaming: bool = False,
     supported_interfaces: Optional[List[Dict[str, Any]]] = None,
     base_url: Optional[str] = None,
+    security_schemes: Optional[Dict[str, Any]] = None,
+    security_requirements: Optional[List[Dict[str, Any]]] = None,
 ) -> Dict[str, Any]:
     """Create or update an external A2A agent discovered from Nacos.
 
@@ -279,6 +302,8 @@
         streaming: Whether this agent supports SSE streaming.
         supported_interfaces: All supported protocol interfaces.
         base_url: Base URL for health checks (service root address).
+        security_schemes: Security schemes declared by the Agent Card.
+        security_requirements: Security requirements declared by the Agent Card.
 
     Returns:
         Created agent information dict.
@@ -309,6 +334,8 @@
             existing.streaming = streaming
             existing.supported_interfaces = supported_interfaces
             existing.raw_card = raw_card
+            existing.security_schemes = security_schemes
+            existing.security_requirements = security_requirements
             existing.cached_at = now
             existing.cache_expires_at = expires_at
             existing.updated_by = user_id
@@ -327,6 +354,8 @@
                 source_type="nacos",
                 nacos_config_id=nacos_config_id,
                 nacos_agent_name=nacos_agent_name,
+                security_schemes=security_schemes,
+                security_requirements=security_requirements,
                 tenant_id=tenant_id,
                 created_by=user_id,
                 updated_by=user_id,
@@ -349,6 +378,9 @@
             "protocol_type": agent.protocol_type,
             "streaming": agent.streaming,
             "supported_interfaces": agent.supported_interfaces,
+            "security_schemes": agent.security_schemes,
+            "security_requirements": agent.security_requirements,
+            "configured_security_scheme_ids": _configured_security_scheme_ids(agent.security_credentials),
             "source_type": agent.source_type,
             "base_url": agent.base_url,
             "is_available": agent.is_available,
@@ -357,12 +389,19 @@
         }
 
 
-def get_external_agent_by_id(external_agent_id: int, tenant_id: str) -> Optional[Dict[str, Any]]:
+def get_external_agent_by_id(
+    external_agent_id: int,
+    tenant_id: str,
+    include_agent_card_headers: bool = False,
+    include_security_credentials: bool = False,
+) -> Optional[Dict[str, Any]]:
     """Get an external agent by its id.
 
     Args:
         external_agent_id: The external agent database ID.
         tenant_id: Tenant ID for isolation.
+        include_agent_card_headers: Include headers for internal Agent Card refresh only.
+        include_security_credentials: Include credentials for internal agent calls only.
 
     Returns:
         Agent information dict or None if not found.
@@ -377,7 +416,7 @@
         if not agent:
             return None
 
-        return {
+        result = {
             "id": agent.id,
             "name": agent.name,
             "description": agent.description,
@@ -386,6 +425,9 @@
             "streaming": agent.streaming,
             "protocol_type": agent.protocol_type,
             "supported_interfaces": agent.supported_interfaces,
+            "security_schemes": agent.security_schemes,
+            "security_requirements": agent.security_requirements,
+            "configured_security_scheme_ids": _configured_security_scheme_ids(agent.security_credentials),
             "source_type": agent.source_type,
             "source_url": agent.source_url,
             "base_url": agent.base_url,
@@ -399,6 +441,11 @@
             "cache_expires_at": agent.cache_expires_at.isoformat() if agent.cache_expires_at else None,
             "create_time": agent.create_time.isoformat() if agent.create_time else None,
         }
+        if include_agent_card_headers:
+            result["agent_card_headers"] = agent.agent_card_headers
+        if include_security_credentials:
+            result["security_credentials"] = agent.security_credentials
+        return result
 
 
 def list_external_agents(
@@ -444,6 +491,9 @@
                 "streaming": agent.streaming,
                 "protocol_type": agent.protocol_type,
                 "supported_interfaces": agent.supported_interfaces,
+                "security_schemes": agent.security_schemes,
+                "security_requirements": agent.security_requirements,
+                "configured_security_scheme_ids": _configured_security_scheme_ids(agent.security_credentials),
                 "source_type": agent.source_type,
                 "source_url": agent.source_url,
                 "base_url": agent.base_url,
@@ -455,6 +505,30 @@
         ]
 
 
+def update_external_agent_security_credentials(
+    external_agent_id: int,
+    tenant_id: str,
+    user_id: str,
+    security_credentials: Dict[str, str],
+) -> bool:
+    """Save configured security credential values for an external agent."""
+    with _get_db_session() as session:
+        agent = session.query(A2AExternalAgent).filter(
+            A2AExternalAgent.id == external_agent_id,
+            A2AExternalAgent.tenant_id == tenant_id,
+            A2AExternalAgent.delete_flag != 'Y'
+        ).first()
+        if not agent:
+            return False
+
+        existing_credentials = dict(agent.security_credentials or {})
+        existing_credentials.update(security_credentials)
+        agent.security_credentials = existing_credentials
+        agent.updated_by = user_id
+        session.flush()
+        return True
+
+
 def delete_external_agent(external_agent_id: int, tenant_id: str) -> bool:
     """Soft delete an external agent.
 
@@ -569,6 +643,8 @@
             "protocol_type": agent.protocol_type,
             "streaming": agent.streaming,
             "supported_interfaces": agent.supported_interfaces,
+            "security_schemes": agent.security_schemes,
+            "security_requirements": agent.security_requirements,
             "source_type": agent.source_type,
             "source_url": agent.source_url,
             "nacos_config_id": agent.nacos_config_id,
@@ -596,6 +672,8 @@
     new_streaming: Optional[bool] = None,
     new_supported_interfaces: Optional[List[Dict[str, Any]]] = None,
     new_protocol_type: Optional[str] = None,
+    new_security_schemes: Optional[Dict[str, Any]] = None,
+    new_security_requirements: Optional[List[Dict[str, Any]]] = None,
 ) -> Optional[Dict[str, Any]]:
     """Refresh the cache for an external agent.
 
@@ -611,6 +689,8 @@
         new_streaming: Updated streaming capability.
         new_supported_interfaces: Updated supported interfaces.
         new_protocol_type: Updated protocol type (JSONRPC, HTTP+JSON, or GRPC).
+        new_security_schemes: Updated Agent Card security schemes.
+        new_security_requirements: Updated Agent Card security requirements.
 
     Returns:
         Updated agent information dict or None if not found.
@@ -642,6 +722,10 @@
             agent.streaming = new_streaming
         if new_supported_interfaces is not None:
             agent.supported_interfaces = new_supported_interfaces
+        if new_security_schemes is not None:
+            agent.security_schemes = new_security_schemes
+        if new_security_requirements is not None:
+            agent.security_requirements = new_security_requirements
         if new_protocol_type is not None:
             agent.protocol_type = new_protocol_type
             # Update agent_url based on the selected protocol type

backend/database/db_models.py

@@ -1568,6 +1568,18 @@ class A2AExternalAgent(TableBase):
 
     # For URL mode
     source_url = Column(String(512), doc="Direct URL to agent card")
+    agent_card_headers = Column(
+        JSON,
+        doc="Headers used only to retrieve and refresh the Agent Card"
+    )
+
+    # Security declared by the Agent Card and credentials configured by the user
+    security_schemes = Column(JSON, doc="Security schemes declared by the Agent Card")
+    security_requirements = Column(JSON, doc="Security requirements declared by the Agent Card")
+    security_credentials = Column(
+        JSON,
+        doc="Credential values for Agent Card security schemes, never exposed by APIs"
+    )
 
     # For Nacos mode
     nacos_config_id = Column(