diff --git a/pyproject.toml b/pyproject.toml index 82d3c108..2e750880 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "sap-cloud-sdk" -version = "0.53.0" +version = "0.53.1" description = "SAP Cloud SDK for Python" readme = "README.md" license = "Apache-2.0" diff --git a/src/sap_cloud_sdk/agentgateway/_customer.py b/src/sap_cloud_sdk/agentgateway/_customer.py index f0ec1020..577bc3dd 100644 --- a/src/sap_cloud_sdk/agentgateway/_customer.py +++ b/src/sap_cloud_sdk/agentgateway/_customer.py @@ -13,6 +13,7 @@ - Gateway handles mTLS externally, SDK uses standard HTTPS """ +import asyncio import json import logging import os @@ -24,6 +25,8 @@ from mcp import ClientSession from mcp.client.streamable_http import streamable_http_client +from sap_cloud_sdk.agentgateway.config import DEFAULT_MAX_CONCURRENT_TASKS + try: from mcp.shared.exceptions import McpError except ImportError: @@ -733,6 +736,7 @@ async def get_mcp_tools_customer( system_token: str, timeout: float, filter: MCPToolFilter | None = None, + max_concurrent_tasks: int = DEFAULT_MAX_CONCURRENT_TASKS, ) -> list[MCPTool]: """List all MCP tools from servers defined in credentials. @@ -745,10 +749,13 @@ async def get_mcp_tools_customer( timeout: HTTP timeout in seconds for MCP server calls. filter: Optional MCPToolFilter narrowing results by tool name or ORD ID. If None or empty, all tools are included. + max_concurrent_tasks: Maximum number of server fetches that run + concurrently. Defaults to 15. Returns: List of MCPTool objects from all servers. """ + start_time = asyncio.get_event_loop().time() f = filter or MCPToolFilter() dependencies = credentials.integration_dependencies @@ -764,9 +771,10 @@ async def get_mcp_tools_customer( logger.info("Discovering tools from %d MCP server(s)", len(dependencies)) - tools: list[MCPTool] = [] + # Fetch all servers concurrently; isolate per-server failures + semaphore = asyncio.Semaphore(max_concurrent_tasks) - for dep in dependencies: + async def _guarded(dep: IntegrationDependency) -> list[MCPTool]: url = _build_mcp_url(credentials.gateway_url, dep.ord_id, dep.global_tenant_id) logger.debug( "Discovering tools from %s (ord_id=%s, gt_id=%s)", @@ -774,21 +782,33 @@ async def get_mcp_tools_customer( dep.ord_id, dep.global_tenant_id, ) + async with semaphore: + return await _list_server_tools(url, system_token, timeout) - try: - server_tools = await _list_server_tools(url, system_token, timeout) - tools.extend(server_tools) - logger.debug("Loaded %d tool(s) from %s", len(server_tools), dep.ord_id) - except Exception as exc: - _log_mcp_server_error(dep.ord_id, exc) + results = await asyncio.gather( + *(_guarded(dep) for dep in dependencies), + return_exceptions=True, + ) + + tools: list[MCPTool] = [] + for dep, result in zip(dependencies, results): + if isinstance(result, BaseException): + _log_mcp_server_error(dep.ord_id, result) + else: + tools.extend(result) + logger.debug("Loaded %d tool(s) from %s", len(result), dep.ord_id) # Post-fetch filter: tool names are only known after fetching if f.names: names_set = set(f.names) tools = [t for t in tools if t.name in names_set] + elapsed = asyncio.get_event_loop().time() - start_time logger.info( - "Loaded %d MCP tool(s) from %d server(s)", len(tools), len(dependencies) + "Loaded %d MCP tool(s) from %d server(s) in %.2fs", + len(tools), + len(dependencies), + elapsed, ) return tools diff --git a/src/sap_cloud_sdk/agentgateway/_lob.py b/src/sap_cloud_sdk/agentgateway/_lob.py index 3813b998..57ec4b8b 100644 --- a/src/sap_cloud_sdk/agentgateway/_lob.py +++ b/src/sap_cloud_sdk/agentgateway/_lob.py @@ -18,6 +18,7 @@ from mcp.shared.exceptions import McpError except ImportError: from mcp.shared.exceptions import MCPError as McpError # type: ignore[no-redef] # ty: ignore[unresolved-import] +from sap_cloud_sdk.agentgateway.config import DEFAULT_MAX_CONCURRENT_TASKS from sap_cloud_sdk.destination import ( create_client as create_destination_client, ConsumptionLevel, @@ -410,6 +411,7 @@ async def get_mcp_tools_lob( system_token: str, timeout: float, filter: MCPToolFilter | None = None, + max_concurrent_tasks: int = DEFAULT_MAX_CONCURRENT_TASKS, ) -> list[MCPTool]: """List all MCP tools using LoB flow (destination-based). @@ -421,10 +423,13 @@ async def get_mcp_tools_lob( timeout: HTTP timeout in seconds for MCP server calls. filter: Optional MCPToolFilter narrowing results by tool name or ORD ID. If None or empty, all tools are included. + max_concurrent_tasks: Maximum number of fragment fetches that run + concurrently. Defaults to 15. Returns: List of MCPTool objects from all MCP servers. """ + start_time = asyncio.get_event_loop().time() f = filter or MCPToolFilter() tools: list[MCPTool] = [] loop = asyncio.get_running_loop() @@ -451,35 +456,53 @@ async def get_mcp_tools_lob( in ord_ids_set ] + # Collect fragments that have a valid URL; skip and warn the rest up front + tasks: list[tuple[str, str]] = [] for fragment in fragments: fragment_name = fragment.name mcp_url = fragment.properties.get("URL") or fragment.properties.get("url") - if not mcp_url: logger.warning( "Fragment '%s' has no URL property — skipping", fragment_name ) continue + tasks.append((fragment_name, mcp_url)) - try: - server_tools = await list_server_tools( + # Fetch all fragments concurrently; isolate per-fragment failures + semaphore = asyncio.Semaphore(max_concurrent_tasks) + + async def _guarded_mcp(mcp_url: str, fragment_name: str) -> list[MCPTool]: + async with semaphore: + return await list_server_tools( mcp_url, system_token, fragment_name, timeout ) - tools.extend(server_tools) + + results = await asyncio.gather( + *(_guarded_mcp(mcp_url, fragment_name) for fragment_name, mcp_url in tasks), + return_exceptions=True, + ) + + for (fragment_name, _), result in zip(tasks, results): + if isinstance(result, BaseException): + _log_mcp_server_error(fragment_name, result) + else: + tools.extend(result) logger.debug( - "Loaded %d tool(s) from fragment '%s'", - len(server_tools), - fragment_name, + "Loaded %d tool(s) from fragment '%s'", len(result), fragment_name ) - except Exception as exc: - _log_mcp_server_error(fragment_name, exc) # Post-fetch filter: tool names are only known after fetching if f.names: names_set = set(f.names) tools = [t for t in tools if t.name in names_set] - logger.info("Loaded %d MCP tool(s) from %d fragment(s)", len(tools), len(fragments)) + elapsed = asyncio.get_event_loop().time() - start_time + logger.info( + "Loaded %d MCP tool(s) from %d fragment(s) in %.2fs", + len(tools), + len(tasks), + elapsed, + ) return tools @@ -610,6 +633,7 @@ async def get_agent_cards_lob( system_token: str, timeout: float, filter: AgentCardFilter | None = None, + max_concurrent_tasks: int = DEFAULT_MAX_CONCURRENT_TASKS, ) -> list[Agent]: """List A2A agents and their agent cards using LoB flow. @@ -628,10 +652,13 @@ async def get_agent_cards_lob( timeout: HTTP timeout in seconds. filter: Optional AgentCardFilter narrowing results by agent card name or ORD ID. If None or empty, all A2A fragments are included. + max_concurrent_tasks: Maximum number of agent card fetches that run + concurrently. Defaults to 15. Returns: List of Agent objects, each containing ORD ID and fetched AgentCard. """ + start_time = asyncio.get_event_loop().time() f = filter or AgentCardFilter() loop = asyncio.get_running_loop() @@ -656,8 +683,8 @@ async def get_agent_cards_lob( in ord_ids_set ] - agents: list[Agent] = [] - + # Collect fragments that have a valid URL and extractable ORD ID; skip the rest + tasks: list[tuple[str, str, str]] = [] for fragment in fragments: fragment_name = fragment.name props_lower = {k.lower(): v for k, v in fragment.properties.items()} @@ -680,14 +707,32 @@ async def get_agent_cards_lob( ) continue - try: - card = await _fetch_agent_card(fragment_url, system_token, timeout) - agents.append(Agent(ord_id=ord_id, agent_card=card)) - logger.debug("Fetched agent card for fragment '%s'", fragment_name) - except Exception: + tasks.append((fragment_name, fragment_url, ord_id)) + + # Fetch all agent cards concurrently; isolate per-fragment failures + semaphore = asyncio.Semaphore(max_concurrent_tasks) + + async def _guarded_card(fragment_url: str) -> AgentCard: + async with semaphore: + return await _fetch_agent_card(fragment_url, system_token, timeout) + + card_results = await asyncio.gather( + *(_guarded_card(fragment_url) for _, fragment_url, _ in tasks), + return_exceptions=True, + ) + elapsed = asyncio.get_event_loop().time() - start_time + + agents: list[Agent] = [] + for (fragment_name, _, ord_id), result in zip(tasks, card_results): + if isinstance(result, BaseException): logger.exception( - "Failed to fetch agent card for fragment '%s' — skipping", fragment_name + "Failed to fetch agent card for fragment '%s' — skipping", + fragment_name, + exc_info=result, ) + else: + agents.append(Agent(ord_id=ord_id, agent_card=result)) + logger.debug("Fetched agent card for fragment '%s'", fragment_name) # Post-fetch filter: agent card name is only known after fetching if f.agent_names: @@ -695,6 +740,9 @@ async def get_agent_cards_lob( agents = [a for a in agents if a.agent_card.raw.get("name") in agent_names_set] logger.info( - "Fetched %d agent card(s) from %d A2A fragment(s)", len(agents), len(fragments) + "Fetched %d agent card(s) from %d A2A fragment(s) in %.2fs", + len(agents), + len(tasks), + elapsed, ) return agents diff --git a/src/sap_cloud_sdk/agentgateway/agw_client.py b/src/sap_cloud_sdk/agentgateway/agw_client.py index 51203dd5..e2b42bc4 100644 --- a/src/sap_cloud_sdk/agentgateway/agw_client.py +++ b/src/sap_cloud_sdk/agentgateway/agw_client.py @@ -439,6 +439,7 @@ async def list_mcp_tools( auth.access_token, self._config.timeout, filter=filter, + max_concurrent_tasks=self._config.max_concurrent_tasks, ) # Check for transparent mode @@ -450,6 +451,7 @@ async def list_mcp_tools( auth.access_token, self._config.timeout, filter=filter, + max_concurrent_tasks=self._config.max_concurrent_tasks, ) # LoB flow - requires tenant_subdomain @@ -459,6 +461,7 @@ async def list_mcp_tools( auth.access_token, self._config.timeout, filter=filter, + max_concurrent_tasks=self._config.max_concurrent_tasks, ) except AgentGatewaySDKError: @@ -525,6 +528,7 @@ async def list_agent_cards( auth.access_token, self._config.timeout, filter=filter, + max_concurrent_tasks=self._config.max_concurrent_tasks, ) except AgentGatewaySDKError: raise diff --git a/src/sap_cloud_sdk/agentgateway/config.py b/src/sap_cloud_sdk/agentgateway/config.py index 17495dbd..1b6589df 100644 --- a/src/sap_cloud_sdk/agentgateway/config.py +++ b/src/sap_cloud_sdk/agentgateway/config.py @@ -7,6 +7,7 @@ DEFAULT_TOKEN_EXPIRY_BUFFER_SECONDS = 30.0 DEFAULT_MAX_SYSTEM_TOKEN_CACHE_SIZE = 32 DEFAULT_MAX_USER_TOKEN_CACHE_SIZE = 256 +DEFAULT_MAX_CONCURRENT_TASKS = 15 @dataclass @@ -22,6 +23,9 @@ class ClientConfig: token expiries before a cached token is considered stale. max_system_token_cache_size: Maximum number of cached system tokens. max_user_token_cache_size: Maximum number of cached user tokens. + max_concurrent_tasks: Maximum number of MCP and Agent fetches (MCP tool + discovery or A2A agent card fetches) that run concurrently. + Defaults to 15. """ timeout: float = DEFAULT_TIMEOUT_SECONDS @@ -29,6 +33,7 @@ class ClientConfig: token_expiry_buffer_seconds: float = DEFAULT_TOKEN_EXPIRY_BUFFER_SECONDS max_system_token_cache_size: int = DEFAULT_MAX_SYSTEM_TOKEN_CACHE_SIZE max_user_token_cache_size: int = DEFAULT_MAX_USER_TOKEN_CACHE_SIZE + max_concurrent_tasks: int = DEFAULT_MAX_CONCURRENT_TASKS def __post_init__(self) -> None: if self.token_expiry_buffer_seconds >= self.fallback_token_ttl_seconds: diff --git a/src/sap_cloud_sdk/agentgateway/user-guide.md b/src/sap_cloud_sdk/agentgateway/user-guide.md index c578bff3..b18cdafa 100644 --- a/src/sap_cloud_sdk/agentgateway/user-guide.md +++ b/src/sap_cloud_sdk/agentgateway/user-guide.md @@ -173,6 +173,12 @@ The SDK discovers resources via BTP Destination Service fragments filtered by th - **Customer flow:** N/A - **Further reading:** N/A +## Performance + +`list_mcp_tools` (LoB and Customer flows) and `list_agent_cards` (LoB only) fetch all fragments **concurrently** using `asyncio.gather`. Wall-clock time is bounded by the slowest single fragment regardless of how many fragments are registered. + +The maximum number of concurrent fetches is controlled by `ClientConfig.max_concurrent_tasks` (default: `15`). Lower it if you need to reduce connection pressure on the gateway or increase it to improve performance when dealing with many fragments. + ## API ### Factory Function @@ -200,6 +206,7 @@ config = ClientConfig( token_expiry_buffer_seconds=30.0, max_system_token_cache_size=32, max_user_token_cache_size=256, + max_concurrent_tasks=15, ) agw_client = create_client(tenant_subdomain="my-tenant", config=config) @@ -210,6 +217,7 @@ agw_client = create_client(tenant_subdomain="my-tenant", config=config) - `token_expiry_buffer_seconds`: Safety buffer subtracted from explicit token expiries before a cached token is reused. Default: `30.0`. - `max_system_token_cache_size`: Maximum number of cached system tokens per client instance. Default: `32`. - `max_user_token_cache_size`: Maximum number of cached exchanged user tokens per client instance. Default: `256`. +- `max_concurrent_tasks`: Maximum number of MCP tool and agent card fetches that run concurrently during `list_mcp_tools` and `list_agent_cards`. Default: `15`. The SDK keeps token caches per `AgentGatewayClient` instance and reuses valid cached tokens for repeated authentication calls. System and user token caches are bounded independently with least-recently-used eviction. diff --git a/tests/agentgateway/unit/test_agw_client.py b/tests/agentgateway/unit/test_agw_client.py index 9fb5dcde..bdf6e70d 100644 --- a/tests/agentgateway/unit/test_agw_client.py +++ b/tests/agentgateway/unit/test_agw_client.py @@ -488,7 +488,7 @@ async def test_with_callable_tenant(self): await agw_client.list_mcp_tools() mock_lob.assert_called_once_with( - "my-tenant", "system-token", 60.0, filter=None + "my-tenant", "system-token", 60.0, filter=None, max_concurrent_tasks=15 ) @pytest.mark.asyncio @@ -515,7 +515,7 @@ async def test_calls_lob_flow_with_system_token(self): await agw_client.list_mcp_tools() mock_lob.assert_called_once_with( - "my-tenant", "system-token-xyz", 60.0, filter=None + "my-tenant", "system-token-xyz", 60.0, filter=None, max_concurrent_tasks=15 ) @pytest.mark.asyncio @@ -581,7 +581,7 @@ async def test_customer_flow_passes_system_token(self): await agw_client.list_mcp_tools() mock_customer.assert_called_once_with( - mock_creds, "customer-system-token", 60.0, filter=None + mock_creds, "customer-system-token", 60.0, filter=None, max_concurrent_tasks=15 ) @pytest.mark.asyncio @@ -609,7 +609,7 @@ async def test_lob_flow_with_user_token_uses_user_auth(self): assert mock_user_auth.call_count == 1 mock_lob.assert_called_once_with( - "my-tenant", "user-token-xyz", 60.0, filter=None + "my-tenant", "user-token-xyz", 60.0, filter=None, max_concurrent_tasks=15 ) @pytest.mark.asyncio @@ -637,7 +637,7 @@ async def test_customer_flow_with_user_token_uses_user_auth(self): await agw_client.list_mcp_tools(user_token="user-jwt") mock_customer.assert_called_once_with( - mock_creds, "exchanged-user-token", 60.0, filter=None + mock_creds, "exchanged-user-token", 60.0, filter=None, max_concurrent_tasks=15 ) @pytest.mark.asyncio @@ -671,6 +671,7 @@ async def test_passes_filter_arguments_lob(self): "token", 60.0, filter=f, + max_concurrent_tasks=15, ) @pytest.mark.asyncio @@ -697,7 +698,7 @@ async def test_empty_filter_passes_through_to_lob(self): await agw_client.list_mcp_tools(filter=f) mock_lob.assert_called_once_with( - "my-tenant", "token", 60.0, filter=f + "my-tenant", "token", 60.0, filter=f, max_concurrent_tasks=15 ) @pytest.mark.asyncio @@ -737,6 +738,7 @@ async def test_passes_filter_arguments_customer(self): "customer-system-token", 60.0, filter=f, + max_concurrent_tasks=15, ) @@ -1047,6 +1049,7 @@ async def test_returns_agents_from_lob_flow(self): "system-token", 60.0, filter=None, + max_concurrent_tasks=15, ) @pytest.mark.asyncio @@ -1080,6 +1083,7 @@ async def test_passes_filter_arguments(self): filter=AgentCardFilter( agent_names=["Billing Agent"], ord_ids=["sap.s4:agent:v1"] ), + max_concurrent_tasks=15, ) @pytest.mark.asyncio diff --git a/tests/agentgateway/unit/test_customer.py b/tests/agentgateway/unit/test_customer.py index 512875be..4a7dac3e 100644 --- a/tests/agentgateway/unit/test_customer.py +++ b/tests/agentgateway/unit/test_customer.py @@ -693,6 +693,80 @@ async def test_filter_excluding_all_dependencies_returns_empty(self): assert result == [] assert mock_list.call_count == 0 + @pytest.mark.asyncio + async def test_servers_fetched_concurrently(self): + """All servers start before any finishes — confirms asyncio.gather parallelism.""" + import asyncio + + active: set[str] = set() + max_concurrent = 0 + + async def slow_fetch(url, token, timeout): + nonlocal max_concurrent + ord_id = url.split("/")[-2] + active.add(ord_id) + max_concurrent = max(max_concurrent, len(active)) + await asyncio.sleep(0.05) + active.discard(ord_id) + return [] + + credentials = CustomerCredentials( + token_service_url="https://ias.example.com/oauth2/token", + client_id="test-client", + certificate="cert", + private_key="key", + gateway_url="https://agw.example.com", + integration_dependencies=[ + IntegrationDependency(ord_id=f"server{i}", global_tenant_id="t") + for i in range(3) + ], + ) + + with patch( + "sap_cloud_sdk.agentgateway._customer._list_server_tools", + side_effect=slow_fetch, + ): + await get_mcp_tools_customer(credentials, "token", 60.0) + + assert max_concurrent > 1, "Expected servers to be fetched concurrently" + + @pytest.mark.asyncio + async def test_fetch_errors_isolated_per_server(self): + """A BaseException from one server does not prevent others from being fetched.""" + mock_tool = MCPTool( + name="tool-ok", + server_name="server2", + description="OK", + input_schema={}, + url="https://example.com", + ) + + async def mock_list(url, token, timeout): + if "server1" in url: + raise RuntimeError("server1 exploded") + return [mock_tool] + + credentials = CustomerCredentials( + token_service_url="https://ias.example.com/oauth2/token", + client_id="test-client", + certificate="cert", + private_key="key", + gateway_url="https://agw.example.com", + integration_dependencies=[ + IntegrationDependency(ord_id="server1", global_tenant_id="t"), + IntegrationDependency(ord_id="server2", global_tenant_id="t"), + ], + ) + + with patch( + "sap_cloud_sdk.agentgateway._customer._list_server_tools", + side_effect=mock_list, + ): + result = await get_mcp_tools_customer(credentials, "token", 60.0) + + assert len(result) == 1 + assert result[0].name == "tool-ok" + # ============================================================ # Test: call_mcp_tool_customer diff --git a/tests/agentgateway/unit/test_lob.py b/tests/agentgateway/unit/test_lob.py index fd9396a9..47d3cc40 100644 --- a/tests/agentgateway/unit/test_lob.py +++ b/tests/agentgateway/unit/test_lob.py @@ -822,6 +822,128 @@ async def test_empty_filter_lists_behave_like_none(self): assert [t.name for t in result] == ["get-sales-order"] + @pytest.mark.asyncio + async def test_fragments_fetched_concurrently(self): + """All fragments are dispatched concurrently, not one-by-one.""" + import asyncio as _asyncio + + started: list[str] = [] + finished: list[str] = [] + + async def slow_tools(url, token, name, timeout): + started.append(name) + await _asyncio.sleep(0.05) + finished.append(name) + return [ + MCPTool( + name=f"tool-{name}", + server_name=name, + description="", + input_schema={}, + url=url, + fragment_name=name, + ) + ] + + fragments = [] + for i in range(3): + f = MagicMock() + f.name = f"frag-{i}" + f.properties = {"URL": f"https://example.com/mcp/{i}"} + fragments.append(f) + + with ( + patch("sap_cloud_sdk.agentgateway._lob.list_mcp_fragments") as mock_list, + patch( + "sap_cloud_sdk.agentgateway._lob.list_server_tools", + side_effect=slow_tools, + ), + ): + mock_list.return_value = fragments + result = await get_mcp_tools_lob("tenant-sub", "token", 60.0) + + # All 3 started before any finished — proves concurrent dispatch + assert len(started) == 3 + assert set(started) == {"frag-0", "frag-1", "frag-2"} + assert len(result) == 3 + + @pytest.mark.asyncio + async def test_all_fragments_attempted_even_if_some_fail(self): + """Failures in some fragments do not prevent others from being fetched.""" + good = MagicMock() + good.name = "good" + good.properties = {"URL": "https://example.com/mcp/good"} + + bad1 = MagicMock() + bad1.name = "bad1" + bad1.properties = {"URL": "https://example.com/mcp/bad1"} + + bad2 = MagicMock() + bad2.name = "bad2" + bad2.properties = {"URL": "https://example.com/mcp/bad2"} + + expected_tool = MCPTool( + name="good-tool", + server_name="good", + description="", + input_schema={}, + url="https://example.com/mcp/good", + fragment_name="good", + ) + + async def selective(*args, **kwargs): + name = args[2] + if name != "good": + raise RuntimeError(f"connection refused: {name}") + return [expected_tool] + + with ( + patch("sap_cloud_sdk.agentgateway._lob.list_mcp_fragments") as mock_list, + patch( + "sap_cloud_sdk.agentgateway._lob.list_server_tools", + side_effect=selective, + ), + ): + mock_list.return_value = [bad1, bad2, good] + result = await get_mcp_tools_lob("tenant-sub", "token", 60.0) + + assert len(result) == 1 + assert result[0].name == "good-tool" + + @pytest.mark.asyncio + async def test_fragment_count_in_log_excludes_url_missing_fragments(self): + """The final 'Loaded N tool(s) from M fragment(s)' counts only fetchable fragments.""" + no_url = MagicMock() + no_url.name = "no-url" + no_url.properties = {} + + with_url = MagicMock() + with_url.name = "with-url" + with_url.properties = {"URL": "https://example.com/mcp"} + + tool = MCPTool( + name="t", + server_name="s", + description="", + input_schema={}, + url="https://example.com/mcp", + fragment_name="with-url", + ) + + with ( + patch("sap_cloud_sdk.agentgateway._lob.list_mcp_fragments") as mock_list, + patch( + "sap_cloud_sdk.agentgateway._lob.list_server_tools", + new_callable=AsyncMock, + return_value=[tool], + ), + ): + mock_list.return_value = [no_url, with_url] + result = await get_mcp_tools_lob("tenant-sub", "token", 60.0) + + # Only the fragment with a URL contributes to results + assert len(result) == 1 + # ============================================================ # Test: list_server_tools @@ -1373,6 +1495,75 @@ async def _selective_fetch(fragment_url, token, timeout): assert len(result) == 1 assert result[0].ord_id == "ord-ok" + @pytest.mark.asyncio + async def test_fragments_fetched_concurrently(self): + """All A2A fragments are dispatched concurrently, not one-by-one.""" + import asyncio as _asyncio + + started: list[str] = [] + + async def slow_fetch(fragment_url, token, timeout): + started.append(fragment_url) + await _asyncio.sleep(0.05) + return AgentCard(raw={"name": "Agent"}) + + fragments = [ + self._make_fragment( + f"frag-{i}", + f"https://agw.example.com/v1/a2a/ord-{i}/tenant", + ) + for i in range(3) + ] + + with ( + patch( + "sap_cloud_sdk.agentgateway._lob.list_a2a_fragments", + return_value=fragments, + ), + patch( + "sap_cloud_sdk.agentgateway._lob._fetch_agent_card", + side_effect=slow_fetch, + ), + ): + result = await get_agent_cards_lob("tenant-sub", "token", 60.0) + + # All 3 started before any finished — proves concurrent dispatch + assert len(started) == 3 + assert len(result) == 3 + + @pytest.mark.asyncio + async def test_fetch_errors_isolated_per_fragment(self): + """A fetch error on one fragment does not abort the others.""" + frags = [ + self._make_fragment( + f"frag-{i}", + f"https://agw.example.com/v1/a2a/ord-{i}/tenant", + ) + for i in range(3) + ] + + async def selective(fragment_url, token, timeout): + if "ord-1" in fragment_url: + raise ConnectionError("timeout") + return AgentCard(raw={"name": "Agent"}) + + with ( + patch( + "sap_cloud_sdk.agentgateway._lob.list_a2a_fragments", + return_value=frags, + ), + patch( + "sap_cloud_sdk.agentgateway._lob._fetch_agent_card", + side_effect=selective, + ), + ): + result = await get_agent_cards_lob("tenant-sub", "token", 60.0) + + # ord-1 failed; ord-0 and ord-2 succeed + assert len(result) == 2 + ord_ids = {a.ord_id for a in result} + assert ord_ids == {"ord-0", "ord-2"} + class TestGetIasClientIdLob: """Tests for get_ias_client_id_lob().""" diff --git a/uv.lock b/uv.lock index 2e7c15f3..f0a1cac5 100644 --- a/uv.lock +++ b/uv.lock @@ -166,9 +166,9 @@ name = "aiologic" version = "0.17.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "sniffio", marker = "python_full_version < '3.13'" }, - { name = "typing-extensions", marker = "python_full_version < '3.13'" }, - { name = "wrapt", marker = "python_full_version < '3.13'" }, + { name = "sniffio" }, + { name = "typing-extensions" }, + { name = "wrapt" }, ] sdist = { url = "https://files.pythonhosted.org/packages/f1/7a/d51f2fde1e8ae8a83431f8e97b7a71e9358cdb1d4d2ce6be387fa44d68de/aiologic-0.17.1.tar.gz", hash = "sha256:2e1b93b9e88ced318c2a63ad7b382688f40cbfe40e3d42258d49dc9c5aea179d", size = 252354, upload-time = "2026-06-27T20:41:33.25Z" } wheels = [ @@ -816,8 +816,8 @@ name = "culsans" version = "0.11.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "aiologic", marker = "python_full_version < '3.13'" }, - { name = "typing-extensions", marker = "python_full_version < '3.13'" }, + { name = "aiologic" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/d9/e3/49afa1bc180e0d28008ec6bcdf82a4072d1c7a41032b5b759b60814ca4b0/culsans-0.11.0.tar.gz", hash = "sha256:0b43d0d05dce6106293d114c86e3fb4bfc63088cfe8ff08ed3fe36891447fe33", size = 107546, upload-time = "2025-12-31T23:15:38.196Z" } wheels = [ @@ -4286,7 +4286,7 @@ wheels = [ [[package]] name = "sap-cloud-sdk" -version = "0.53.0" +version = "0.53.1" source = { editable = "." } dependencies = [ { name = "cryptography" },