diff --git a/backend/agents/create_agent_info.py b/backend/agents/create_agent_info.py index d58932946..98cde314f 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 @@ -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"), ) diff --git a/backend/apps/a2a_client_app.py b/backend/apps/a2a_client_app.py index ea149ac31..7d2b7cb35 100644 --- a/backend/apps/a2a_client_app.py +++ b/backend/apps/a2a_client_app.py @@ -46,6 +46,16 @@ class UpdateAgentProtocolRequest(BaseModel): ) +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 @@ 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, + 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, diff --git a/backend/database/a2a_agent_db.py b/backend/database/a2a_agent_db.py index c1d998272..a063f7503 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, + "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, diff --git a/backend/database/db_models.py b/backend/database/db_models.py index b0d6d710f..26588fcf8 100644 --- a/backend/database/db_models.py +++ b/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") diff --git a/backend/services/a2a_client_service.py b/backend/services/a2a_client_service.py index e4e81fec5..7cfe382d5 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, + 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: 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..4b4e202b1 --- /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 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'; diff --git a/frontend/app/[locale]/agents/components/a2a/A2AAgentDiscoveryModal.tsx b/frontend/app/[locale]/agents/components/a2a/A2AAgentDiscoveryModal.tsx index bc9260a29..2ded3d572 100644 --- a/frontend/app/[locale]/agents/components/a2a/A2AAgentDiscoveryModal.tsx +++ b/frontend/app/[locale]/agents/components/a2a/A2AAgentDiscoveryModal.tsx @@ -88,28 +88,53 @@ 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; + timeoutSeconds: number | 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 [timeoutSeconds, setTimeoutSeconds] = useState(String(agent.timeout_seconds || 300)); + 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]); + setTimeoutSeconds(String(agent.timeout_seconds || 300)); + setSettingsError(null); + }, [(agent as any).protocol_type, agent.timeout_seconds]); + + const handleSave = async () => { + setSettingsError(null); + const parsedTimeout = Number(timeoutSeconds); + + try { + if (!Number.isInteger(parsedTimeout) || parsedTimeout < 1 || parsedTimeout > 3600) { + throw new Error(t("a2a.protocol.timeoutRange", { defaultValue: "Timeout must be between 1 and 3600 seconds" })); + } + } 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, + timeoutSeconds: parsedTimeout, + }); setSaving(false); - setOpen(false); + if (saved) setOpen(false); }; return ( @@ -147,6 +172,24 @@ function AgentProtocolSetting({ agent, onProtocolChange }: Readonly +
+ + {t("a2a.protocol.timeoutSeconds", { defaultValue: "Timeout seconds" })} + + setTimeoutSeconds(e.target.value)} + /> +
+ {settingsError && ( + + {settingsError} + + )}