From ec8c12f85637a062317a9940568eb4812f8fd598 Mon Sep 17 00:00:00 2001 From: xuyaqist Date: Mon, 27 Jul 2026 21:00:17 +0800 Subject: [PATCH 1/7] Feat: support a2a custom authorization --- backend/apps/a2a_client_app.py | 76 +++++++- backend/consts/a2a_models.py | 6 +- backend/database/a2a_agent_db.py | 88 ++++++++- backend/database/db_models.py | 12 ++ backend/services/a2a_client_service.py | 155 +++++++++++++--- backend/utils/a2a_http_client.py | 32 +++- ...agent_card_headers_and_security_fields.sql | 17 ++ doc/docs/en/user-guide/agent-development.md | 14 +- doc/docs/zh/user-guide/agent-development.md | 14 +- .../components/a2a/A2AAgentDiscoveryModal.tsx | 147 ++++++++++++++- frontend/public/locales/en/common.json | 12 ++ frontend/public/locales/zh/common.json | 12 ++ frontend/services/a2aService.ts | 32 ++++ frontend/services/api.ts | 2 + test/backend/consts/test_a2a_models.py | 19 ++ test/backend/database/test_a2a_agent_db.py | 82 ++++++++- .../services/test_a2a_client_service.py | 172 +++++++++++++++++- test/backend/utils/test_a2a_http_client.py | 14 ++ 18 files changed, 844 insertions(+), 62 deletions(-) create mode 100644 deploy/sql/migrations/v2.4.0_0727_add_a2a_agent_card_headers_and_security_fields.sql diff --git a/backend/apps/a2a_client_app.py b/backend/apps/a2a_client_app.py index ea149ac31..e78c8bee3 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 @@ -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 UpdateAgentProtocolRequest(BaseModel): ) +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 @@ async def discover_from_url( 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 @@ async def get_external_agent( ) +@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, diff --git a/backend/consts/a2a_models.py b/backend/consts/a2a_models.py index 29eb10414..2ec309006 100644 --- a/backend/consts/a2a_models.py +++ b/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): diff --git a/backend/database/a2a_agent_db.py b/backend/database/a2a_agent_db.py index c1d998272..69bf765fa 100644 --- a/backend/database/a2a_agent_db.py +++ b/backend/database/a2a_agent_db.py @@ -145,6 +145,11 @@ def _extract_protocol_type(supported_interfaces: Optional[List[Dict[str, Any]]]) 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, @@ -157,6 +162,9 @@ def create_external_agent_from_url( 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 @@ def create_external_agent_from_url( 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 @@ def create_external_agent_from_url( 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 @@ def create_external_agent_from_url( 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 @@ def create_external_agent_from_url( "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, @@ -263,6 +284,8 @@ def create_external_agent_from_nacos( 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 @@ def create_external_agent_from_nacos( 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 @@ def create_external_agent_from_nacos( 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 @@ def create_external_agent_from_nacos( 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 @@ def create_external_agent_from_nacos( "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 create_external_agent_from_nacos( } -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 @@ def get_external_agent_by_id(external_agent_id: int, tenant_id: str) -> Optional if not agent: return None - return { + result = { "id": agent.id, "name": agent.name, "description": agent.description, @@ -386,6 +425,9 @@ def get_external_agent_by_id(external_agent_id: int, tenant_id: str) -> Optional "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 @@ def get_external_agent_by_id(external_agent_id: int, tenant_id: str) -> Optional "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 @@ def list_external_agents( "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 list_external_agents( ] +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 @@ def update_external_agent_protocol( "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 @@ def refresh_external_agent_cache( 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 @@ def refresh_external_agent_cache( 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 @@ def refresh_external_agent_cache( 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 diff --git a/backend/database/db_models.py b/backend/database/db_models.py index aec452479..bf86cf930 100644 --- a/backend/database/db_models.py +++ b/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( diff --git a/backend/services/a2a_client_service.py b/backend/services/a2a_client_service.py index e4e81fec5..23c6da900 100644 --- a/backend/services/a2a_client_service.py +++ b/backend/services/a2a_client_service.py @@ -6,12 +6,13 @@ """ import asyncio import logging +from typing import Any, AsyncIterator, Dict, List, Optional, Tuple + import aiohttp -from typing import Any, AsyncIterator, Dict, List, Optional from database import a2a_agent_db from database.a2a_agent_db import _extract_protocol_type, PROTOCOL_HTTP_JSON, PROTOCOL_JSONRPC -from utils.a2a_http_client import A2AHttpClient, build_a2a_headers +from utils.a2a_http_client import A2AHttpClient, A2AHttpStatusError, build_a2a_headers logger = logging.getLogger(__name__) @@ -45,7 +46,8 @@ async def discover_from_url( self, url: str, tenant_id: str, - user_id: str + user_id: str, + custom_headers: Optional[Dict[str, str]] = None ) -> Dict[str, Any]: """Discover an external A2A agent from a URL. @@ -55,6 +57,7 @@ async def discover_from_url( url: Direct URL to the Agent Card (e.g., https://example.com/.well-known/agent-xxx.json). tenant_id: Tenant ID for isolation. user_id: User who initiated the discovery. + custom_headers: Headers saved only for Agent Card discovery and refresh. Returns: Discovered agent information dict. @@ -63,8 +66,11 @@ async def discover_from_url( AgentDiscoveryError: If discovery fails. """ try: + custom_headers = custom_headers or {} async with A2AHttpClient() as client: headers = build_a2a_headers() + if custom_headers: + headers.update(custom_headers) card = await client.get_json(url, headers=headers) # Extract agent info from Card @@ -117,7 +123,10 @@ async def discover_from_url( tenant_id=tenant_id, user_id=user_id, raw_card=card, - supported_interfaces=supported_interfaces + supported_interfaces=supported_interfaces, + agent_card_headers=custom_headers, + security_schemes=card.get("securitySchemes"), + security_requirements=card.get("securityRequirements"), ) logger.info(f"Discovered A2A agent {agent_id} from URL: {url}") @@ -320,7 +329,9 @@ async def _discover_single_from_nacos( tenant_id=tenant_id, user_id=user_id, raw_card=agent_info, - supported_interfaces=supported_interfaces + supported_interfaces=supported_interfaces, + security_schemes=agent_info.get("securitySchemes"), + security_requirements=agent_info.get("securityRequirements"), ) return result @@ -473,7 +484,11 @@ async def refresh_agent_card( AgentDiscoveryError: If refresh fails. """ # Get current agent info - agent = a2a_agent_db.get_external_agent_by_id(external_agent_id, tenant_id) + agent = a2a_agent_db.get_external_agent_by_id( + external_agent_id, + tenant_id, + include_agent_card_headers=True, + ) if not agent: raise AgentDiscoveryError(f"Agent {external_agent_id} not found") @@ -524,7 +539,11 @@ async def refresh_agent_card( raise AgentDiscoveryError("No source URL available for refresh") async with A2AHttpClient() as client: - card = await client.get_json(source_url) + card_headers = agent.get("agent_card_headers") + headers = build_a2a_headers() + if isinstance(card_headers, dict): + headers.update(card_headers) + card = await client.get_json(source_url, headers=headers) # Extract updated info - use _extract_agent_url for A2A v1.0 standard new_url = self._extract_agent_url(card) @@ -559,7 +578,9 @@ async def refresh_agent_card( new_name=new_name, new_description=new_description, new_supported_interfaces=new_supported_interfaces, - new_protocol_type=new_protocol_type + new_protocol_type=new_protocol_type, + new_security_schemes=card.get("securitySchemes", {}), + new_security_requirements=card.get("securityRequirements", []), ) elif update_agent_url: # Only agent_url changed @@ -575,7 +596,9 @@ async def refresh_agent_card( new_agent_url=new_url, new_name=new_name, new_description=new_description, - new_supported_interfaces=new_supported_interfaces + new_supported_interfaces=new_supported_interfaces, + new_security_schemes=card.get("securitySchemes", {}), + new_security_requirements=card.get("securityRequirements", []), ) else: # No changes to agent_url or protocol_type, just update metadata @@ -586,7 +609,9 @@ async def refresh_agent_card( new_raw_card=card, new_name=new_name, new_description=new_description, - new_supported_interfaces=new_supported_interfaces + new_supported_interfaces=new_supported_interfaces, + new_security_schemes=card.get("securitySchemes", {}), + new_security_requirements=card.get("securityRequirements", []), ) # Update availability @@ -627,15 +652,10 @@ def delete_external_agent(self, external_agent_id: int, tenant_id: str) -> bool: # ============================================================================= def _build_endpoint_url(self, agent_url: str, protocol_type: str, streaming: bool = False) -> str: - """Build the complete endpoint URL by appending protocol-specific path. + """Build the request URL from the Agent Card protocol binding. - Args: - agent_url: Base agent URL from database. - protocol_type: Protocol type (JSONRPC, HTTP+JSON, GRPC). - streaming: Whether this is a streaming request. - - Returns: - Complete endpoint URL with protocol path appended. + The Agent Card URL is a complete endpoint for JSON-RPC. HTTP+JSON URLs + identify the service base and require the A2A message operation suffix. """ base_url = agent_url.rstrip("/") path_suffix = self._get_protocol_path(protocol_type, streaming) @@ -644,13 +664,51 @@ def _build_endpoint_url(self, agent_url: str, protocol_type: str, streaming: boo return base_url def _get_protocol_path(self, protocol_type: str, streaming: bool) -> str: - """Get the path suffix for a given protocol type and streaming mode.""" + """Get the required HTTP+JSON operation suffix for an A2A request.""" if protocol_type == PROTOCOL_HTTP_JSON: return "/message:stream" if streaming else "/message:send" - if protocol_type == PROTOCOL_JSONRPC: - return "/v1" return "" + def _build_security_request_parts( + self, + agent: Dict[str, Any], + ) -> Tuple[Dict[str, str], Dict[str, str], Dict[str, str]]: + """Build request authentication parts from the configured Card security scheme.""" + schemes = agent.get("security_schemes") or {} + requirements = agent.get("security_requirements") or [] + credentials = agent.get("security_credentials") or {} + if not requirements: + return {}, {}, {} + + for requirement in requirements: + required_schemes = requirement.get("schemes", {}) if isinstance(requirement, dict) else {} + if not required_schemes: + return {}, {}, {} + + headers: Dict[str, str] = {} + params: Dict[str, str] = {} + cookies: Dict[str, str] = {} + valid = True + for scheme_id in required_schemes: + credential = credentials.get(scheme_id) + scheme = schemes.get(scheme_id, {}) + api_key_scheme = scheme.get("apiKeySecurityScheme", {}) if isinstance(scheme, dict) else {} + location = api_key_scheme.get("location") + parameter_name = api_key_scheme.get("name") + if not credential or not parameter_name or location not in {"header", "query", "cookie"}: + valid = False + break + if location == "header": + headers[parameter_name] = credential + elif location == "query": + params[parameter_name] = credential + else: + cookies[parameter_name] = credential + if valid: + return headers, params, cookies + + raise AgentCallError("Configured credentials do not satisfy the Agent Card security requirements") + async def call_agent( self, external_agent_id: int, @@ -672,7 +730,11 @@ async def call_agent( AgentCallError: If the call fails. """ # Get agent info - agent = a2a_agent_db.get_external_agent_by_id(external_agent_id, tenant_id) + agent = a2a_agent_db.get_external_agent_by_id( + external_agent_id, + tenant_id, + include_security_credentials=True, + ) if not agent: raise AgentCallError(f"Agent {external_agent_id} not found") @@ -685,7 +747,10 @@ async def call_agent( # Build complete endpoint URL with protocol path endpoint_url = self._build_endpoint_url(agent_url, protocol_type, streaming=False) - logger.info(f"[A2A-CLIENT] === Calling external A2A agent === id={external_agent_id}, url={endpoint_url}, protocol={protocol_type}, message={message}") + logger.info( + f"[A2A-CLIENT] Calling external agent id={external_agent_id}, " + f"url={endpoint_url}, protocol={protocol_type}" + ) try: # Build request based on protocol type @@ -705,11 +770,22 @@ async def call_agent( "message": message } - logger.info(f"Calling external A2A agent {external_agent_id}: url={endpoint_url}, protocol={protocol_type}, payload={payload}") + logger.info( + f"Calling external A2A agent {external_agent_id}: " + f"url={endpoint_url}, protocol={protocol_type}" + ) headers = build_a2a_headers() + auth_headers, auth_params, auth_cookies = self._build_security_request_parts(agent) + headers.update(auth_headers) async with A2AHttpClient() as client: - response = await client.post_json(endpoint_url, payload, headers) + response = await client.post_json( + endpoint_url, + payload, + headers, + params=auth_params, + cookies=auth_cookies, + ) # Parse response if "error" in response: @@ -718,6 +794,14 @@ async def call_agent( return response.get("result", response) + except A2AHttpStatusError as e: + logger.error(f"External agent {external_agent_id} returned HTTP {e.status}") + if e.status == 401: + raise AgentCallError( + "External agent authentication failed (HTTP 401). " + "Check the configured security credentials." + ) from e + raise AgentCallError(f"External agent request failed (HTTP {e.status})") from e except aiohttp.ClientError as e: logger.error(f"Failed to call agent {external_agent_id}: {e}") raise AgentCallError(f"Call failed: {str(e)}") from e @@ -744,7 +828,11 @@ async def call_agent_streaming( AgentCallError: If the call setup fails. """ # Get agent info - agent = a2a_agent_db.get_external_agent_by_id(external_agent_id, tenant_id) + agent = a2a_agent_db.get_external_agent_by_id( + external_agent_id, + tenant_id, + include_security_credentials=True, + ) if not agent: raise AgentCallError(f"Agent {external_agent_id} not found") @@ -774,13 +862,24 @@ async def call_agent_streaming( "message": message } - logger.info(f"Calling external A2A agent {external_agent_id} (streaming): url={endpoint_url}, protocol={protocol_type}, payload={payload}") + logger.info( + f"Calling external A2A agent {external_agent_id} (streaming): " + f"url={endpoint_url}, protocol={protocol_type}" + ) headers = build_a2a_headers(api_key) + auth_headers, auth_params, auth_cookies = self._build_security_request_parts(agent) + headers.update(auth_headers) try: async with A2AHttpClient() as client: - async for event in client.post_stream(endpoint_url, payload, headers): + async for event in client.post_stream( + endpoint_url, + payload, + headers, + params=auth_params, + cookies=auth_cookies, + ): yield event except aiohttp.ClientError as e: logger.error(f"Streaming call failed for agent {external_agent_id}: {e}") diff --git a/backend/utils/a2a_http_client.py b/backend/utils/a2a_http_client.py index 8b7c55d9f..b5c03c4e9 100644 --- a/backend/utils/a2a_http_client.py +++ b/backend/utils/a2a_http_client.py @@ -22,6 +22,14 @@ CONTENT_TYPE_JSON = "application/json" +class A2AHttpStatusError(Exception): + """Raised when an A2A endpoint returns a non-success HTTP status.""" + + def __init__(self, method: str, url: str, status: int): + super().__init__(f"A2A {method} request to {url} failed with HTTP {status}") + self.status = status + + class A2AHttpClient: """HTTP client for A2A protocol communication.""" @@ -147,6 +155,9 @@ async def get_json( url, headers=request_headers ) + if not 200 <= status < 300: + raise A2AHttpStatusError("GET", url, status) + # Decode body and handle empty responses body_text = body.decode('utf-8') if body else "" @@ -179,7 +190,9 @@ async def post_json( self, url: str, payload: Dict[str, Any], - headers: Optional[Dict[str, str]] = None + headers: Optional[Dict[str, str]] = None, + params: Optional[Dict[str, str]] = None, + cookies: Optional[Dict[str, str]] = None, ) -> Dict[str, Any]: """Send a POST request and return JSON response.""" if not self._session: @@ -195,15 +208,20 @@ async def post_json( if headers: request_headers.update(headers) - logger.info(f"A2A POST request: url={url}, payload={payload}") + logger.info(f"A2A POST request: url={url}") try: status, body = await self._request_with_retry( "POST", url, json=payload, - headers=request_headers + headers=request_headers, + params=params, + cookies=cookies, ) + if not 200 <= status < 300: + raise A2AHttpStatusError("POST", url, status) + # Decode body and handle empty responses body_text = body.decode('utf-8') if body else "" @@ -240,7 +258,9 @@ async def post_stream( self, url: str, payload: Dict[str, Any], - headers: Optional[Dict[str, str]] = None + headers: Optional[Dict[str, str]] = None, + params: Optional[Dict[str, str]] = None, + cookies: Optional[Dict[str, str]] = None, ) -> AsyncIterator[Dict[str, Any]]: """Send a streaming POST request and yield SSE events.""" if not self._session: @@ -250,7 +270,9 @@ async def post_stream( response = await self._session.post( url, json=payload, - headers=headers + headers=headers, + params=params, + cookies=cookies, ) response.raise_for_status() diff --git a/deploy/sql/migrations/v2.4.0_0727_add_a2a_agent_card_headers_and_security_fields.sql b/deploy/sql/migrations/v2.4.0_0727_add_a2a_agent_card_headers_and_security_fields.sql new file mode 100644 index 000000000..4014553ba --- /dev/null +++ b/deploy/sql/migrations/v2.4.0_0727_add_a2a_agent_card_headers_and_security_fields.sql @@ -0,0 +1,17 @@ +ALTER TABLE nexent.ag_a2a_external_agent_t + ADD COLUMN IF NOT EXISTS agent_card_headers JSONB; + +COMMENT ON COLUMN nexent.ag_a2a_external_agent_t.agent_card_headers + IS 'Headers saved only for Agent Card discovery and refresh'; + +ALTER TABLE nexent.ag_a2a_external_agent_t + ADD COLUMN IF NOT EXISTS security_schemes JSONB, + ADD COLUMN IF NOT EXISTS security_requirements JSONB, + ADD COLUMN IF NOT EXISTS security_credentials JSONB; + +COMMENT ON COLUMN nexent.ag_a2a_external_agent_t.security_schemes + IS 'Security schemes declared by the Agent Card'; +COMMENT ON COLUMN nexent.ag_a2a_external_agent_t.security_requirements + IS 'Security requirements declared by the Agent Card'; +COMMENT ON COLUMN nexent.ag_a2a_external_agent_t.security_credentials + IS 'Credential values for Agent Card security schemes, never exposed by APIs'; diff --git a/doc/docs/en/user-guide/agent-development.md b/doc/docs/en/user-guide/agent-development.md index 8e6b47d4f..3e309712f 100644 --- a/doc/docs/en/user-guide/agent-development.md +++ b/doc/docs/en/user-guide/agent-development.md @@ -61,9 +61,12 @@ If you know the Agent Card address of the target agent, you can use the URL disc 1. In the External A2A Agent list, click the "Add External Agent" button 2. Select the "URL Discovery" tab 3. Fill in the Agent Card URL address, for example: `https://example.com/.well-known/agent.json` -4. Click the "Discover" button; the system will automatically retrieve the agent's related information -5. After successful discovery, you can view the agent's name, description, capabilities and other information -6. Click "Add to List" to complete the addition +4. If the target Agent Card requires authentication, enter a JSON object in "Custom Request Headers", for example: `{"Authorization": "Bearer "}` +5. Click the "Discover" button; the system will automatically retrieve the agent's related information +6. After successful discovery, you can view the agent's name, description, capabilities and other information +7. Click "Add to List" to complete the addition + +> 💡 **Tip**: Custom request headers are saved with the external agent and used only to retrieve and refresh its Agent Card. They are never used for agent calls. When rediscovering the same URL, leaving this field empty keeps the current configuration; entering `{}` clears it. > 💡 **Tip**: The Agent Card is an Agent description file that complies with the A2A 1.0 specification, containing the agent's name, description, calling address, capabilities and other information. @@ -103,8 +106,9 @@ In the External A2A Agent list, you can view and manage all discovered external 4. **Configure Calling Protocol**: Click the "Protocol Configuration" button to select the calling protocol for this agent: - **HTTP + JSON**: Use REST API style calls - **JSON-RPC**: Use JSON-RPC protocol calls -5. **Refresh Agent Information**: If the agent information changes, click the "Refresh" button to re-fetch the latest Agent Card -6. **Remove Agent**: Click the "Remove" button to delete the agent from the discovered list +5. **Configure Call Authentication**: If the Agent Card declares `securitySchemes` and `securityRequirements`, click "Agent Authentication" and enter the required values. Nexent places each value in the header, query string, or cookie specified by the Card; fields in the same requirement must all be configured. +6. **Refresh Agent Information**: If the agent information changes, click the "Refresh" button to re-fetch the latest Agent Card +7. **Remove Agent**: Click the "Remove" button to delete the agent from the discovered list > 💡 **Use Cases**: > - Quickly integrate known third-party agent services through URL discovery diff --git a/doc/docs/zh/user-guide/agent-development.md b/doc/docs/zh/user-guide/agent-development.md index 40805aeea..f38813f3a 100644 --- a/doc/docs/zh/user-guide/agent-development.md +++ b/doc/docs/zh/user-guide/agent-development.md @@ -61,9 +61,12 @@ Nexent 支持通过 A2A 协议与第三方 Agent 进行通信。您可以通过 1. 在外部 A2A Agent 列表中，点击"添加外部 Agent"按钮 2. 选择"URL 发现"页签 3. 填写 Agent Card URL 地址，例如：`https://example.com/.well-known/agent.json` -4. 点击"发现"按钮，系统会自动获取 Agent 的相关信息 -5. 发现成功后，可以查看 Agent 的名称、描述、能力等信息 -6. 点击"添加到列表"完成添加 +4. 如果目标 Agent Card 需要认证，在"自定义请求头"中填写 JSON 对象，例如：`{"Authorization": "Bearer "}` +5. 点击"发现"按钮，系统会自动获取 Agent 的相关信息 +6. 发现成功后，可以查看 Agent 的名称、描述、能力等信息 +7. 点击"添加到列表"完成添加 + +> 💡 **提示**：自定义请求头会随该外部 Agent 保存，仅用于获取和刷新 Agent Card，不会用于后续调用 Agent。再次发现同一 URL 时，留空会保留现有配置，填写 `{}` 可清空配置。 > 💡 **提示**：Agent Card 是符合 A2A 1.0 规范的 Agent 描述文件，包含了 Agent 的名称、描述、调用地址、能力等信息。 @@ -105,8 +108,9 @@ Nexent 支持通过 A2A 协议与第三方 Agent 进行通信。您可以通过 4. **配置调用协议**：点击"协议配置"按钮，可以选择该 Agent 的调用协议： - **HTTP + JSON**：使用 REST API 风格调用 - **JSON-RPC**：使用 JSON-RPC 协议调用 -5. **刷新 Agent 信息**：如果 Agent 信息发生变化，可以点击"刷新"按钮重新获取最新的 Agent Card -6. **移除 Agent**：点击"移除"按钮，可以将该 Agent 从已发现列表中删除 +5. **配置调用认证**：如果 Agent Card 声明了 `securitySchemes` 和 `securityRequirements`，点击"Agent 认证"按钮，填写所需认证值。系统会按 Card 声明将值放入请求头、查询参数或 Cookie；同一认证组合中的字段必须同时填写。 +6. **刷新 Agent 信息**：如果 Agent 信息发生变化，可以点击"刷新"按钮重新获取最新的 Agent Card +7. **移除 Agent**：点击"移除"按钮，可以将该 Agent 从已发现列表中删除 > 💡 **使用场景**： > - 通过 URL 发现快速接入已知的第三方 Agent 服务 diff --git a/frontend/app/[locale]/agents/components/a2a/A2AAgentDiscoveryModal.tsx b/frontend/app/[locale]/agents/components/a2a/A2AAgentDiscoveryModal.tsx index bc9260a29..f3f5b16a6 100644 --- a/frontend/app/[locale]/agents/components/a2a/A2AAgentDiscoveryModal.tsx +++ b/frontend/app/[locale]/agents/components/a2a/A2AAgentDiscoveryModal.tsx @@ -31,6 +31,7 @@ import { Search, Eye, Settings, + KeyRound, MessageCircle, } from "lucide-react"; import { a2aClientService, A2AExternalAgent } from "@/services/a2aService"; @@ -58,6 +59,114 @@ const PROTOCOL_BINDING_MAP: Record = { "grpc": "GRPC", }; +interface AgentSecuritySettingProps { + agent: A2AExternalAgent; + onSaved: () => void; +} + +function AgentSecuritySetting({ agent, onSaved }: Readonly) { + const { t } = useTranslation("common"); + const [open, setOpen] = useState(false); + const [values, setValues] = useState>({}); + const [saving, setSaving] = useState(false); + const [configuredSchemeIds, setConfiguredSchemeIds] = useState( + agent.configured_security_scheme_ids || [], + ); + const schemes = agent.security_schemes || {}; + const requirements = agent.security_requirements || []; + const entries = Object.entries(schemes).flatMap(([schemeId, scheme]) => { + const apiKeyScheme = (scheme as Record).apiKeySecurityScheme; + return apiKeyScheme ? [{ schemeId, ...apiKeyScheme }] : []; + }); + + useEffect(() => { + setConfiguredSchemeIds(agent.configured_security_scheme_ids || []); + }, [agent.configured_security_scheme_ids]); + + if (entries.length === 0 || requirements.length === 0) { + return null; + } + + const handleSave = async () => { + const configuredValues = Object.fromEntries( + Object.entries(values).filter(([, value]) => value.trim()), + ); + if (Object.keys(configuredValues).length === 0) { + message.error(t("a2a.security.valueRequired")); + return; + } + + setSaving(true); + const result = await a2aClientService.updateAgentSecurityCredentials( + String(agent.id), + configuredValues, + ); + setSaving(false); + if (result.success) { + setConfiguredSchemeIds(result.data?.configured_security_scheme_ids || configuredSchemeIds); + message.success(t("a2a.security.saveSuccess")); + setValues({}); + setOpen(false); + onSaved(); + } else { + message.error(result.message || t("a2a.security.saveFailed")); + } + }; + + return ( + + + {t("a2a.security.requirementsHint")} + + + {entries.map(({ schemeId, description, location, name }) => { + const isConfigured = configuredSchemeIds.includes(schemeId); + return ( +
+ + {name || schemeId} + {location && {location}} + {isConfigured && ( + {t("a2a.security.configured")} + )} + + setValues((current) => ({ + ...current, + [schemeId]: event.target.value, + }))} + /> +
+ ); + })} + +
+ + + + +
+ + } + title={t("a2a.security.settings")} + trigger="click" + open={open} + onOpenChange={setOpen} + > + +