diff --git a/backend/agents/create_agent_info.py b/backend/agents/create_agent_info.py index d58932946..51273b4d1 100644 --- a/backend/agents/create_agent_info.py +++ b/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"), ) diff --git a/backend/apps/a2a_client_app.py b/backend/apps/a2a_client_app.py index ea149ac31..4de35866e 100644 --- a/backend/apps/a2a_client_app.py +++ b/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 UpdateAgentProtocolRequest(BaseModel): ) +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 @@ async def delete_external_agent( ) +@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, diff --git a/backend/database/a2a_agent_db.py b/backend/database/a2a_agent_db.py index c1d998272..da8aa59cd 100644 --- a/backend/database/a2a_agent_db.py +++ b/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, diff --git a/backend/database/db_models.py b/backend/database/db_models.py index b0d6d710f..94e8904fe 100644 --- a/backend/database/db_models.py +++ b/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") diff --git a/backend/services/a2a_client_service.py b/backend/services/a2a_client_service.py index e4e81fec5..fe65afd7c 100644 --- a/backend/services/a2a_client_service.py +++ b/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: diff --git a/backend/utils/a2a_http_client.py b/backend/utils/a2a_http_client.py index 8b7c55d9f..ccc0ac328 100644 --- a/backend/utils/a2a_http_client.py +++ b/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 diff --git a/deploy/sql/migrations/v2.3.0_0702_add_a2a_call_settings.sql b/deploy/sql/migrations/v2.3.0_0702_add_a2a_call_settings.sql new file mode 100644 index 000000000..67cd7f7fa --- /dev/null +++ b/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'; diff --git a/frontend/app/[locale]/agents/components/a2a/A2AAgentDiscoveryModal.tsx b/frontend/app/[locale]/agents/components/a2a/A2AAgentDiscoveryModal.tsx index bc9260a29..cbdcbbaf3 100644 --- a/frontend/app/[locale]/agents/components/a2a/A2AAgentDiscoveryModal.tsx +++ b/frontend/app/[locale]/agents/components/a2a/A2AAgentDiscoveryModal.tsx @@ -88,28 +88,65 @@ function extractAvailableProtocols(supportedInterfaces?: Record[]): // Agent Protocol Setting Popover Component interface AgentProtocolSettingProps { agent: A2AExternalAgent; - onProtocolChange: (agentId: string, protocolType: string) => void; + onSettingsChange: ( + agentId: string, + settings: { + protocolType: string; + customHeaders: Record | null; + } + ) => Promise; } -function AgentProtocolSetting({ agent, onProtocolChange }: Readonly) { +function AgentProtocolSetting({ agent, onSettingsChange }: Readonly) { const { t } = useTranslation("common"); const [open, setOpen] = useState(false); const [selectedProtocol, setSelectedProtocol] = useState( (agent as any).protocol_type || "JSONRPC" ); + const [customHeaders, setCustomHeaders] = useState( + agent.custom_headers ? JSON.stringify(agent.custom_headers, null, 2) : "" + ); + const [settingsError, setSettingsError] = useState(null); const [saving, setSaving] = useState(false); const availableProtocols = extractAvailableProtocols(agent.supported_interfaces); useEffect(() => { setSelectedProtocol((agent as any).protocol_type || "JSONRPC"); - }, [(agent as any).protocol_type]); + setCustomHeaders(agent.custom_headers ? JSON.stringify(agent.custom_headers, null, 2) : ""); + setSettingsError(null); + }, [(agent as any).protocol_type, agent.custom_headers]); + + const parseCustomHeaders = (): Record | null => { + const trimmed = customHeaders.trim(); + if (!trimmed) return null; + const parsed = JSON.parse(trimmed); + if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) { + throw new Error(t("a2a.protocol.headersJsonObject", { defaultValue: "Headers must be a JSON object" })); + } + return Object.fromEntries( + Object.entries(parsed).map(([key, value]) => [key, String(value)]) + ); + }; + + const handleSave = async () => { + setSettingsError(null); + let parsedHeaders: Record | null; + + try { + parsedHeaders = parseCustomHeaders(); + } catch (error) { + setSettingsError(error instanceof Error ? error.message : t("a2a.protocol.invalidSettings", { defaultValue: "Invalid settings" })); + return; + } - const handleSave = () => { setSaving(true); - onProtocolChange(String(agent.id), selectedProtocol); + const saved = await onSettingsChange(String(agent.id), { + protocolType: selectedProtocol, + customHeaders: parsedHeaders, + }); setSaving(false); - setOpen(false); + if (saved) setOpen(false); }; return ( @@ -147,6 +184,22 @@ function AgentProtocolSetting({ agent, onProtocolChange }: Readonly +
+ + {t("a2a.protocol.customHeadersJson", { defaultValue: "Custom headers JSON" })} + + setCustomHeaders(e.target.value)} + placeholder={'{"Authorization":"Bearer token"}'} + /> +
+ {settingsError && ( + + {settingsError} + + )}