fix: configure timeout 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
@@ -470,7 +470,7 @@ def _build_external_agent_config(agent: dict, agent_url: str) -> ExternalA2AAgen
         transport_type=agent.get("transport_type", "http-streaming"),
         protocol_version=agent.get("protocol_version", "1.0"),
         protocol_type=agent.get("protocol_type", PROTOCOL_JSONRPC),
-        timeout=300.0,
+        timeout=float(agent.get("timeout_seconds") or 300),
         raw_card=agent.get("raw_card"),
     )
 

backend/apps/a2a_client_app.py

@@ -46,6 +46,16 @@
     )
 
 
+class UpdateAgentCallSettingsRequest(BaseModel):
+    """Request to update call settings for an external A2A agent."""
+    timeout_seconds: Optional[int] = Field(
+        default=None,
+        ge=1,
+        le=3600,
+        description="Request timeout in seconds for 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 +289,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,
+            timeout_seconds=request.timeout_seconds,
+        )
+
+        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,
+            "timeout_seconds": getattr(agent, "timeout_seconds", 300),
             "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,
+            "timeout_seconds": getattr(agent, "timeout_seconds", 300),
             "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,
+            "timeout_seconds": getattr(agent, "timeout_seconds", 300),
             "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,
+                "timeout_seconds": getattr(agent, "timeout_seconds", 300),
                 "supported_interfaces": agent.supported_interfaces,
                 "source_type": agent.source_type,
                 "source_url": agent.source_url,
@@ -519,6 +523,65 @@ def _find_interface_by_protocol_type(
     return None
 
 
+def _validate_timeout_seconds(timeout_seconds: Optional[int]) -> Optional[int]:
+    """Validate A2A timeout settings."""
+    if timeout_seconds is None:
+        return None
+    if timeout_seconds < 1 or timeout_seconds > 3600:
+        raise ValueError("timeout_seconds must be between 1 and 3600")
+    return timeout_seconds
+
+
+def update_external_agent_call_settings(
+    external_agent_id: int,
+    tenant_id: str,
+    user_id: str,
+    timeout_seconds: Optional[int] = None,
+) -> Optional[Dict[str, Any]]:
+    """Update custom call settings for an external A2A agent."""
+    validated_timeout = _validate_timeout_seconds(timeout_seconds)
+
+    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 timeout_seconds is not None:
+            setattr(agent, "timeout_seconds", validated_timeout)
+        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,
+            "timeout_seconds": getattr(agent, "timeout_seconds", 300),
+            "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 +630,7 @@ def update_external_agent_protocol(
             "version": agent.version,
             "agent_url": agent.agent_url,
             "protocol_type": agent.protocol_type,
+            "timeout_seconds": getattr(agent, "timeout_seconds", 300),
             "streaming": agent.streaming,
             "supported_interfaces": agent.supported_interfaces,
             "source_type": agent.source_type,
@@ -852,6 +916,7 @@ def query_external_sub_agents(
                 "version": agent.version,
                 "agent_url": agent.agent_url,
                 "protocol_type": agent.protocol_type,
+                "timeout_seconds": getattr(agent, "timeout_seconds", 300),
                 "streaming": agent.streaming,
                 "supported_interfaces": agent.supported_interfaces,
                 "raw_card": agent.raw_card,

backend/database/db_models.py

@@ -1137,6 +1137,9 @@ class A2AExternalAgent(TableBase):
     protocol_type = Column(String(20), default=PROTOCOL_JSONRPC,
                            doc="Protocol type for calling this agent")
 
+    timeout_seconds = Column(Integer, default=300,
+                             doc="Request timeout in seconds 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,
+        timeout_seconds: Optional[int] = 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,
+            timeout_seconds=timeout_seconds,
+        )
+
     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)
+        timeout_seconds = float(agent.get("timeout_seconds") or 300)
 
         # Build complete endpoint URL with protocol path
         endpoint_url = self._build_endpoint_url(agent_url, protocol_type, streaming=False)
@@ -708,7 +724,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()
-            async with A2AHttpClient() as client:
+            async with A2AHttpClient(timeout=timeout_seconds) as client:
                 response = await client.post_json(endpoint_url, payload, headers)
 
             # Parse response
@@ -753,6 +769,7 @@ async def call_agent_streaming(
 
         agent_url = agent["agent_url"]
         protocol_type = agent.get("protocol_type", PROTOCOL_JSONRPC)
+        timeout_seconds = float(agent.get("timeout_seconds") or 300)
 
         # Build complete endpoint URL with protocol path
         endpoint_url = self._build_endpoint_url(agent_url, protocol_type, streaming=True)
@@ -779,7 +796,7 @@ async def call_agent_streaming(
         headers = build_a2a_headers(api_key)
 
         try:
-            async with A2AHttpClient() as client:
+            async with A2AHttpClient(timeout=timeout_seconds) as client:
                 async for event in client.post_stream(endpoint_url, payload, headers):
                     yield event
         except aiohttp.ClientError as e:

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 timeout_seconds INTEGER DEFAULT 300;
+
+COMMENT ON COLUMN nexent.ag_a2a_external_agent_t.timeout_seconds IS 'Request timeout in seconds for calling this external A2A agent';