fix: support custom headers for A2A agents

backend/agents/create_agent_info.py

@@ -1,4 +1,4 @@
-import json
+import json
 import threading
 import logging
 from typing import Any, Dict, List, Optional
@@ -471,6 +471,7 @@ def _build_external_agent_config(agent: dict, agent_url: str) -> ExternalA2AAgen
         protocol_version=agent.get("protocol_version", "1.0"),
         protocol_type=agent.get("protocol_type", PROTOCOL_JSONRPC),
         timeout=300.0,
+        custom_headers=agent.get("custom_headers"),
         raw_card=agent.get("raw_card"),
     )
 

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
@@ -46,6 +46,14 @@
     )
 
 
+class UpdateAgentCallSettingsRequest(BaseModel):
+    """Request to update call settings for an external A2A agent."""
+    custom_headers: Optional[Dict[str, str]] = Field(
+        default=None,
+        description="Custom HTTP headers to include when calling the agent"
+    )
+
+
 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)")
@@ -279,6 +287,51 @@
         )
 
 
+@router.put("/agents/{external_agent_id}/settings")
+async def update_agent_call_settings(
+    external_agent_id: int,
+    request: UpdateAgentCallSettingsRequest,
+    authorization: Annotated[Optional[str], Header()] = None,
+    http_request: Request = None
+):
+    """Update custom call settings for an external A2A agent."""
+    try:
+        user_id, tenant_id, _ = get_current_user_info(authorization, http_request)
+
+        result = a2a_client_service.update_agent_call_settings(
+            external_agent_id=external_agent_id,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            custom_headers=request.custom_headers,
+        )
+
+        if not result:
+            raise HTTPException(
+                status_code=HTTPStatus.NOT_FOUND,
+                detail=f"Agent {external_agent_id} not found"
+            )
+
+        return JSONResponse(
+            status_code=HTTPStatus.OK,
+            content={"status": "success", "data": result}
+        )
+
+    except HTTPException:
+        raise
+    except ValueError as e:
+        logger.error(f"Invalid A2A call settings: {e}")
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST,
+            detail=str(e)
+        )
+    except Exception as e:
+        logger.error(f"Update agent call settings failed: {e}", exc_info=True)
+        raise HTTPException(
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
+            detail="Failed to update agent call settings"
+        )
+
+
 @router.put("/agents/{external_agent_id}/protocol")
 async def update_agent_protocol(
     external_agent_id: int,

backend/database/a2a_agent_db.py

@@ -240,6 +240,7 @@ def create_external_agent_from_url(
             "version": agent.version,
             "agent_url": agent.agent_url,
             "protocol_type": agent.protocol_type,
+            "custom_headers": getattr(agent, "custom_headers", None),
             "streaming": agent.streaming,
             "supported_interfaces": agent.supported_interfaces,
             "source_type": agent.source_type,
@@ -347,6 +348,7 @@ def create_external_agent_from_nacos(
             "version": agent.version,
             "agent_url": agent.agent_url,
             "protocol_type": agent.protocol_type,
+            "custom_headers": getattr(agent, "custom_headers", None),
             "streaming": agent.streaming,
             "supported_interfaces": agent.supported_interfaces,
             "source_type": agent.source_type,
@@ -385,6 +387,7 @@ def get_external_agent_by_id(external_agent_id: int, tenant_id: str) -> Optional
             "agent_url": agent.agent_url,
             "streaming": agent.streaming,
             "protocol_type": agent.protocol_type,
+            "custom_headers": getattr(agent, "custom_headers", None),
             "supported_interfaces": agent.supported_interfaces,
             "source_type": agent.source_type,
             "source_url": agent.source_url,
@@ -443,6 +446,7 @@ def list_external_agents(
                 "agent_url": agent.agent_url,
                 "streaming": agent.streaming,
                 "protocol_type": agent.protocol_type,
+                "custom_headers": getattr(agent, "custom_headers", None),
                 "supported_interfaces": agent.supported_interfaces,
                 "source_type": agent.source_type,
                 "source_url": agent.source_url,
@@ -519,6 +523,74 @@ def _find_interface_by_protocol_type(
     return None
 
 
+def _validate_custom_headers(custom_headers: Optional[Dict[str, Any]]) -> Optional[Dict[str, str]]:
+    """Validate A2A custom headers."""
+    if custom_headers is None:
+        return None
+    if not isinstance(custom_headers, dict):
+        raise ValueError("custom_headers must be an object")
+
+    sanitized = {}
+    for key, value in custom_headers.items():
+        header_name = str(key).strip()
+        if not header_name:
+            raise ValueError("custom header names cannot be empty")
+        if any(ch in header_name for ch in "\r\n:"):
+            raise ValueError(f"invalid custom header name: {header_name}")
+        sanitized[header_name] = str(value)
+    return sanitized
+
+
+def update_external_agent_call_settings(
+    external_agent_id: int,
+    tenant_id: str,
+    user_id: str,
+    custom_headers: Optional[Dict[str, Any]] = None,
+) -> Optional[Dict[str, Any]]:
+    """Update custom call settings for an external A2A agent."""
+    sanitized_headers = _validate_custom_headers(custom_headers)
+
+    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 None
+
+        if custom_headers is not None:
+            setattr(agent, "custom_headers", sanitized_headers)
+        agent.updated_by = user_id
+        agent.update_time = datetime.now(timezone.utc)
+        session.flush()
+
+        return {
+            "id": agent.id,
+            "name": agent.name,
+            "description": agent.description,
+            "version": agent.version,
+            "agent_url": agent.agent_url,
+            "protocol_type": agent.protocol_type,
+            "custom_headers": getattr(agent, "custom_headers", None),
+            "streaming": agent.streaming,
+            "supported_interfaces": agent.supported_interfaces,
+            "source_type": agent.source_type,
+            "source_url": agent.source_url,
+            "nacos_config_id": agent.nacos_config_id,
+            "nacos_agent_name": agent.nacos_agent_name,
+            "raw_card": agent.raw_card,
+            "is_available": agent.is_available,
+            "last_check_at": agent.last_check_at.isoformat() if agent.last_check_at else None,
+            "last_check_result": agent.last_check_result,
+            "cached_at": agent.cached_at.isoformat() if agent.cached_at else None,
+            "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,
+            "update_time": agent.update_time.isoformat() if agent.update_time else None,
+        }
+
+
 def update_external_agent_protocol(
     external_agent_id: int,
     tenant_id: str,
@@ -567,6 +639,7 @@ def update_external_agent_protocol(
             "version": agent.version,
             "agent_url": agent.agent_url,
             "protocol_type": agent.protocol_type,
+            "custom_headers": getattr(agent, "custom_headers", None),
             "streaming": agent.streaming,
             "supported_interfaces": agent.supported_interfaces,
             "source_type": agent.source_type,
@@ -852,6 +925,7 @@ def query_external_sub_agents(
                 "version": agent.version,
                 "agent_url": agent.agent_url,
                 "protocol_type": agent.protocol_type,
+                "custom_headers": getattr(agent, "custom_headers", None),
                 "streaming": agent.streaming,
                 "supported_interfaces": agent.supported_interfaces,
                 "raw_card": agent.raw_card,

backend/database/db_models.py

@@ -1137,6 +1137,8 @@ class A2AExternalAgent(TableBase):
     protocol_type = Column(String(20), default=PROTOCOL_JSONRPC,
                            doc="Protocol type for calling this agent")
 
+    custom_headers = Column(JSON, doc="Custom HTTP headers for calling this agent")
+
     # Capabilities
     streaming = Column(Boolean, default=False,
                        doc="Whether this agent supports SSE streaming")

backend/services/a2a_client_service.py

@@ -426,6 +426,21 @@ def list_external_agents(
             is_available=is_available
         )
 
+    def update_agent_call_settings(
+        self,
+        external_agent_id: int,
+        tenant_id: str,
+        user_id: str,
+        custom_headers: Optional[Dict[str, Any]] = None,
+    ) -> Optional[Dict[str, Any]]:
+        """Update custom call settings for an external agent."""
+        return a2a_agent_db.update_external_agent_call_settings(
+            external_agent_id=external_agent_id,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            custom_headers=custom_headers,
+        )
+
     def update_agent_protocol(
         self,
         external_agent_id: int,
@@ -681,6 +696,7 @@ async def call_agent(
 
         agent_url = agent["agent_url"]
         protocol_type = agent.get("protocol_type", PROTOCOL_JSONRPC)
+        custom_headers = agent.get("custom_headers")
 
         # Build complete endpoint URL with protocol path
         endpoint_url = self._build_endpoint_url(agent_url, protocol_type, streaming=False)
@@ -707,7 +723,7 @@ async def call_agent(
 
             logger.info(f"Calling external A2A agent {external_agent_id}: url={endpoint_url}, protocol={protocol_type}, payload={payload}")
 
-            headers = build_a2a_headers()
+            headers = build_a2a_headers(custom_headers=custom_headers)
             async with A2AHttpClient() as client:
                 response = await client.post_json(endpoint_url, payload, headers)
 
@@ -753,6 +769,7 @@ async def call_agent_streaming(
 
         agent_url = agent["agent_url"]
         protocol_type = agent.get("protocol_type", PROTOCOL_JSONRPC)
+        custom_headers = agent.get("custom_headers")
 
         # Build complete endpoint URL with protocol path
         endpoint_url = self._build_endpoint_url(agent_url, protocol_type, streaming=True)
@@ -776,7 +793,7 @@ async def call_agent_streaming(
 
         logger.info(f"Calling external A2A agent {external_agent_id} (streaming): url={endpoint_url}, protocol={protocol_type}, payload={payload}")
 
-        headers = build_a2a_headers(api_key)
+        headers = build_a2a_headers(api_key, custom_headers=custom_headers)
 
         try:
             async with A2AHttpClient() as client:

backend/utils/a2a_http_client.py

@@ -276,7 +276,10 @@ async def post_stream(
             raise
 
 
-def build_a2a_headers(api_key: Optional[str] = None) -> Dict[str, str]:
+def build_a2a_headers(
+    api_key: Optional[str] = None,
+    custom_headers: Optional[Dict[str, str]] = None
+) -> Dict[str, str]:
     """Build HTTP headers for A2A requests."""
     headers = {
         "Content-Type": CONTENT_TYPE_JSON,
@@ -285,4 +288,6 @@ def build_a2a_headers(api_key: Optional[str] = None) -> Dict[str, str]:
     }
     if api_key:
         headers["Authorization"] = f"Bearer {api_key}"
+    if custom_headers:
+        headers.update(custom_headers)
     return headers

deploy/sql/migrations/v2.3.0_0702_add_a2a_call_settings.sql

@@ -0,0 +1,4 @@
+ALTER TABLE nexent.ag_a2a_external_agent_t
+ADD COLUMN IF NOT EXISTS custom_headers JSONB;
+
+COMMENT ON COLUMN nexent.ag_a2a_external_agent_t.custom_headers IS 'Custom HTTP headers for calling this external A2A agent';