perf(agentgateway): parallelize MCP tools and A2A agent card fetching

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"

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,31 +771,44 @@ 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)",
             url,
             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
 

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,21 +707,42 @@ 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:
         agent_names_set = set(f.agent_names)
         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

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

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,13 +23,17 @@ 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
     fallback_token_ttl_seconds: float = DEFAULT_FALLBACK_TOKEN_TTL_SECONDS
     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:

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.