feat(agentgateway): add global id filter for lob agents

src/sap_cloud_sdk/agentgateway/__init__.py

@@ -60,6 +60,7 @@
     AgentCard,
     AgentCardFilter,
 )
+from sap_cloud_sdk.agentgateway._fragments import ActiveIntegration
 from sap_cloud_sdk.agentgateway.config import ClientConfig
 from sap_cloud_sdk.agentgateway.agw_client import create_client, AgentGatewayClient
 from sap_cloud_sdk.agentgateway.exceptions import (
@@ -83,6 +84,8 @@
     "Agent",
     "AgentCard",
     "AgentCardFilter",
+    # Integration metadata
+    "ActiveIntegration",
     # Exceptions
     "AgentGatewaySDKError",
     "AgentGatewayServerError",

src/sap_cloud_sdk/agentgateway/_fragments.py

@@ -4,16 +4,19 @@
 - Label constants for managed-runtime fragment types
 - Fragment listing by label (MCP, A2A, IAS)
 - IAS fragment name lookup for auth flows
+- Active integration listing for tenant context
 """
 
 import logging
 from enum import Enum
+from typing import Optional, TypedDict
 
 from sap_cloud_sdk.destination import (
     create_fragment_client,
     Label,
     ListOptions,
 )
+from sap_cloud_sdk.destination._models import Level
 
 from sap_cloud_sdk.agentgateway.exceptions import MCPServerNotFoundError
 from sap_cloud_sdk.core.telemetry import Module
@@ -23,6 +26,11 @@
 # Shared label key for all managed-runtime fragment types
 LABEL_KEY = "sap-managed-runtime-type"
 
+# Label keys for integration metadata stored on system fragments
+_LABEL_GTID = "sap-managed-runtime-gtid"
+_LABEL_ORD_ID = "sap-managed-runtime-ordid"
+_LABEL_SYSTEM_TYPE = "sap-managed-runtime-system-type"
+
 _DESTINATION_INSTANCE = "default"
 
 
@@ -35,28 +43,45 @@ class FragmentLabel(str, Enum):
     IAS_USER = "subscriber.ias.user"
 
 
-def _list_fragments_by_label(label: FragmentLabel, tenant_subdomain: str) -> list:
+def _list_fragments_by_label(
+    label: FragmentLabel,
+    tenant_subdomain: str,
+    global_tenant_ids: list[str] | None = None,
+) -> list:
+    filter_labels = [Label(key=LABEL_KEY, values=[label.value])]
+    if global_tenant_ids:
+        filter_labels.append(Label(key=_LABEL_GTID, values=global_tenant_ids))
     client = create_fragment_client(
         instance=_DESTINATION_INSTANCE,
         _telemetry_source=Module.AGENTGATEWAY,
     )
     return client.list_instance_fragments(
-        filter=ListOptions(filter_labels=[Label(key=LABEL_KEY, values=[label.value])]),
+        filter=ListOptions(filter_labels=filter_labels),
         tenant=tenant_subdomain,
     )
 
 
-def list_mcp_fragments(tenant_subdomain: str) -> list:
+def list_mcp_fragments(
+    tenant_subdomain: str,
+    global_tenant_ids: list[str] | None = None,
+) -> list:
     """List destination fragments with MCP server label.
 
     Args:
         tenant_subdomain: Tenant subdomain for multi-tenant lookup.
+        global_tenant_ids: Optional list of global tenant IDs of integrated
+            systems to filter by. When set, only fragments whose
+            ``sap-managed-runtime-gtid`` label matches one of these values are
+            returned (filter is applied server-side by the Destination Service).
 
     Returns:
-        List of fragments with sap-managed-runtime-type=agw.mcp.server label.
+        List of fragments with sap-managed-runtime-type=agw.mcp.server label
+        (and, if provided, matching one of the requested global tenant IDs).
     """
     logger.debug("Fetching MCP fragments for tenant '%s'", tenant_subdomain)
-    return _list_fragments_by_label(FragmentLabel.MCP, tenant_subdomain)
+    return _list_fragments_by_label(
+        FragmentLabel.MCP, tenant_subdomain, global_tenant_ids
+    )
 
 
 def list_a2a_fragments(tenant_subdomain: str) -> list:
@@ -118,3 +143,80 @@ def get_ias_user_fragment_name(tenant_subdomain: str) -> str:
             f"for tenant '{tenant_subdomain}'"
         )
     return fragments[0].name
+
+
+class ActiveIntegration(TypedDict):
+    """Metadata for a connected backend system integration."""
+
+    global_tenant_id: str
+    system_type: Optional[str]
+    integration_dependency: str
+
+
+def _list_active_integrations(tenant_subdomain: str) -> list[ActiveIntegration]:
+    """List all active backend system integrations for the given tenant.
+
+    Reads Destination Service instance fragments to discover active backend
+    system integrations for the given tenant. Each fragment represents a
+    connected backend system (e.g. SAP PCE, SAP S/4HANA).
+
+    Retrieves integration metadata from fragment labels:
+        - sap-managed-runtime-gtid: GTID of the connected partner system.
+        - sap-managed-runtime-system-type: Application namespace (e.g. "sap.pce").
+        - sap-managed-runtime-ordid: Sanitized ORD ID of the integration dependency.
+
+    Args:
+        tenant_subdomain: Subscriber tenant subdomain.
+
+    Returns:
+        List of ActiveIntegration dicts, each with keys:
+            - global_tenant_id: GTID of the connected partner system.
+            - system_type: Application namespace of the partner (e.g. "sap.pce").
+            - integration_dependency: ORD ID of the integration dependency fulfilled.
+        Returns empty list if no active integrations exist.
+    """
+    client = create_fragment_client(
+        instance=_DESTINATION_INSTANCE,
+        _telemetry_source=Module.AGENTGATEWAY,
+    )
+    fragments = client.list_instance_fragments(
+        filter=ListOptions(
+            filter_labels=[
+                Label(
+                    key=LABEL_KEY,
+                    values=[FragmentLabel.MCP.value, FragmentLabel.A2A.value],
+                )
+            ]
+        ),
+        tenant=tenant_subdomain,
+    )
+
+    result: list[ActiveIntegration] = []
+    for fragment in fragments:
+        labels = {
+            lbl.key: lbl.values[0] if lbl.values else None
+            for lbl in client.get_fragment_labels(
+                name=fragment.name,
+                level=Level.SERVICE_INSTANCE,
+                tenant=tenant_subdomain,
+            )
+        }
+        gtid = labels.get(_LABEL_GTID)
+        system_type = labels.get(_LABEL_SYSTEM_TYPE)
+        ord_id = labels.get(_LABEL_ORD_ID)
+
+        if not system_type:
+            logger.debug(
+                "Fragment '%s' is missing system_type label; system_type will be None in result",
+                fragment.name,
+            )
+
+        result.append(
+            ActiveIntegration(
+                global_tenant_id=gtid,
+                system_type=system_type,
+                integration_dependency=ord_id,
+            )
+        )
+
+    return result

src/sap_cloud_sdk/agentgateway/_lob.py

@@ -378,8 +378,11 @@ async def get_mcp_tools_lob(
         tenant_subdomain: Tenant subdomain for multi-tenant lookup.
         system_token: Pre-fetched raw system token (from get_system_auth).
         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.
+        filter: Optional MCPToolFilter narrowing results by tool name, ORD ID,
+            or global tenant ID. If None or empty, all tools are included.
+            ``global_tenant_ids`` filters fragments server-side via the
+            Destination Service. ``ord_ids`` filters before fetching.
+            ``names`` filters after fetching.
 
     Returns:
         List of MCPTool objects from all MCP servers.
@@ -390,7 +393,9 @@ async def get_mcp_tools_lob(
 
     logger.info("Listing MCP fragments for tenant '%s'", tenant_subdomain)
 
-    fragments = await loop.run_in_executor(None, list_mcp_fragments, tenant_subdomain)
+    fragments = await loop.run_in_executor(
+        None, list_mcp_fragments, tenant_subdomain, f.global_tenant_ids or None
+    )
 
     if not fragments:
         logger.debug(

src/sap_cloud_sdk/agentgateway/_models.py

@@ -169,6 +169,13 @@ class MCPToolFilter:
             agents, or matched against IntegrationDependency.ord_id for
             customer agents). Applied before fetching, skipping non-matching
             fragments.
+        global_tenant_ids: Global tenant IDs of the integrated systems whose
+            tools should be listed. Only supported in the LoB flow, where each
+            MCP fragment carries a ``sap-managed-runtime-gtid`` label written
+            by SPII at provisioning time. When set, the Destination Service
+            filters fragments server-side. Ignored by the customer flow (which
+            already scopes tools by the ``integrationDependencies`` in the
+            credentials file).
 
     Example:
         ```python
@@ -178,10 +185,12 @@ class MCPToolFilter:
             filter=MCPToolFilter(
                 names=["get-sales-order"],
                 ord_ids=["sap.s4:apiAccess:salesOrder:v1"],
+                global_tenant_ids=["9e88a0c4-ab32-46d8-b1d3-07cbcac11831"],
             )
         )
         ```
     """
 
     names: list[str] = field(default_factory=list)
     ord_ids: list[str] = field(default_factory=list)
+    global_tenant_ids: list[str] = field(default_factory=list)

src/sap_cloud_sdk/agentgateway/agw_client.py

@@ -39,6 +39,8 @@
 )
 from sap_cloud_sdk.agentgateway._token_cache import _GatewayUrlCache, _TokenCache
 from sap_cloud_sdk.agentgateway.exceptions import AgentGatewaySDKError
+from sap_cloud_sdk.agentgateway import _fragments
+from sap_cloud_sdk.agentgateway._fragments import ActiveIntegration
 from sap_cloud_sdk.core.telemetry import Module, Operation, record_metrics
 
 logger = logging.getLogger(__name__)
@@ -376,8 +378,9 @@ async def list_mcp_tools(
             user_token: User's JWT for principal propagation.
                 Can be a string or a callable returning a string.
                 If provided, uses user-scoped auth instead of system auth.
-            filter: Optional filter to narrow results by tool name or ORD ID.
-                If None or empty, all tools are included.
+            filter: Optional filter to narrow results by tool name, ORD ID, or
+                global tenant ID. If None or empty, all tools are included.
+                See :class:`MCPToolFilter` for supported fields.
 
         Returns:
             List of MCPTool objects from all MCP servers.
@@ -400,6 +403,7 @@ async def list_mcp_tools(
                 filter=MCPToolFilter(
                     names=["get-sales-order"],
                     ord_ids=["sap.s4:apiAccess:salesOrder:v1"],
+                    global_tenant_ids=["<gtid>"],
                 )
             )
             ```
@@ -519,6 +523,36 @@ async def list_agent_cards(
             logger.exception("Unexpected error during agent card discovery")
             raise AgentGatewaySDKError(f"Agent card discovery failed: {e}") from e
 
+    @record_metrics(Module.AGENTGATEWAY, Operation.AGENTGATEWAY_LIST_ACTIVE_INTEGRATIONS)
+    def list_active_integrations(self) -> list[ActiveIntegration]:
+        """List all active backend system integrations for the current tenant.
+
+        Returns the connected backend systems (e.g. SAP PCE, SAP S/4HANA) that
+        are currently active for this tenant. Use this to determine which systems
+        are connected and which GTIDs to pass when loading MCP tools.
+
+        Requires tenant_subdomain to be configured on the client.
+
+        Returns:
+            List of dicts, each with:
+                - global_tenant_id: GTID of the connected partner system.
+                - system_type: Application namespace (e.g. "sap.pce", "sap.s4").
+                - integration_dependency: ORD ID fulfilled by this integration.
+            Returns empty list if no active integrations exist.
+
+        Raises:
+            AgentGatewaySDKError: If tenant_subdomain is not configured.
+
+        Example:
+            ```python
+            integrations = agw_client.list_active_integrations()
+            for i in integrations:
+                print(i["system_type"], i["global_tenant_id"])
+            ```
+        """
+        tenant = self._resolve_tenant_subdomain()
+        return _fragments._list_active_integrations(tenant)
+
     @record_metrics(Module.AGENTGATEWAY, Operation.AGENTGATEWAY_CALL_MCP_TOOL)
     async def call_mcp_tool(
         self,

src/sap_cloud_sdk/core/telemetry/operation.py

@@ -193,6 +193,7 @@ class Operation(str, Enum):
     AGENTGATEWAY_GET_USER_AUTH = "get_user_auth"
     AGENTGATEWAY_LIST_AGENT_CARDS = "list_agent_cards"
     AGENTGATEWAY_GET_IAS_CLIENT_ID = "get_ias_client_id"
+    AGENTGATEWAY_LIST_ACTIVE_INTEGRATIONS = "list_active_integrations"
 
     # Agent Memory Operations
     AGENT_MEMORY_ADD_MEMORY = "add_memory"

tests/agentgateway/unit/test_agw_client.py

@@ -447,6 +447,73 @@ async def test_with_callable_tenant(self):
                 "my-tenant", "system-token", 60.0, filter=None
             )
 
+    @pytest.mark.asyncio
+    async def test_forwards_global_tenant_ids_from_filter_to_lob(self):
+        """MCPToolFilter.global_tenant_ids should reach get_mcp_tools_lob."""
+        with (
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials",
+                return_value=None,
+            ),
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.detect_transparent_credentials",
+                return_value=False,
+            ),
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.fetch_system_auth",
+                new_callable=AsyncMock,
+                return_value=("system-token", "https://agw.example.com"),
+            ),
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.get_mcp_tools_lob",
+                new_callable=AsyncMock,
+                return_value=[],
+            ) as mock_lob,
+        ):
+            agw_client = create_client(tenant_subdomain="my-tenant")
+
+            await agw_client.list_mcp_tools(
+                filter=MCPToolFilter(global_tenant_ids=["gtid-a", "gtid-b"]),
+            )
+
+            mock_lob.assert_called_once_with(
+                "my-tenant",
+                "system-token",
+                60.0,
+                filter=MCPToolFilter(global_tenant_ids=["gtid-a", "gtid-b"]),
+            )
+
+    @pytest.mark.asyncio
+    async def test_empty_filter_is_equivalent_to_no_filter(self):
+        """MCPToolFilter() with no fields set should not restrict results."""
+        with (
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials",
+                return_value=None,
+            ),
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.detect_transparent_credentials",
+                return_value=False,
+            ),
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.fetch_system_auth",
+                new_callable=AsyncMock,
+                return_value=("system-token", "https://agw.example.com"),
+            ),
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.get_mcp_tools_lob",
+                new_callable=AsyncMock,
+                return_value=[],
+            ) as mock_lob,
+        ):
+            agw_client = create_client(tenant_subdomain="my-tenant")
+
+            await agw_client.list_mcp_tools(filter=MCPToolFilter())
+
+            mock_lob.assert_called_once_with(
+                "my-tenant", "system-token", 60.0, filter=MCPToolFilter()
+            )
+
     @pytest.mark.asyncio
     async def test_calls_lob_flow_with_system_token(self):
         """list_mcp_tools should call LoB flow with system token."""