Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 93 additions & 0 deletions src/sap_cloud_sdk/agentgateway/_fragments.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,12 @@
- 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

from sap_cloud_sdk.destination import (
create_fragment_client,
Expand All @@ -25,6 +27,9 @@

_DESTINATION_INSTANCE = "default"

# URL mode path segments used by system integration fragments
_INTEGRATION_URL_MODES = ("mcp", "a2a")


class FragmentLabel(str, Enum):
"""Label values for the sap-managed-runtime-type fragment label key."""
Expand Down Expand Up @@ -118,3 +123,91 @@ def get_ias_user_fragment_name(tenant_subdomain: str) -> str:
f"for tenant '{tenant_subdomain}'"
)
return fragments[0].name


def list_active_integrations(tenant_subdomain: str) -> list[dict]:
"""List all active backend system integrations for the given tenant.

Reads Destination Service instance fragments written by the Destinations
Facilitator during UCL Formation assignment (SPII flow). Each fragment
represents a connected backend system (e.g. SAP PCE, SAP S/4HANA).

Extracts integration details from the fragment URL, which always has the form:
{agw_base_url}/v1/mcp/{ord_id}/{gtid} (MCP integrations)
{agw_base_url}/v1/a2a/{ord_id}/{gtid} (A2A integrations)

Args:
tenant_subdomain: Subscriber tenant subdomain.

Returns:
List of 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 = []
for fragment in fragments:
url = fragment.properties.get("URL", "")
entry = _parse_integration_from_url(url)
if entry is not None:
result.append(entry)
return result


def _parse_integration_from_url(url: str) -> Optional[dict]:
"""Extract integration metadata from a system fragment URL.

Fragment URLs have the form:
{base}/v1/{mode}/{ord_id}/{gtid}
where mode is "mcp" or "a2a", ord_id may contain colons and slashes,
and gtid is the last path segment.

Args:
url: The fragment URL property value.

Returns:
Dict with global_tenant_id, system_type, integration_dependency,
or None if the URL does not match the expected pattern.
"""
parts = url.rstrip("/").split("/")

mode_idx = None
for i, part in enumerate(parts):
if i > 0 and parts[i - 1] == "v1" and part in _INTEGRATION_URL_MODES:
mode_idx = i
break

if mode_idx is None or mode_idx + 2 > len(parts) - 1:
logger.debug("Skipping fragment with unexpected URL pattern: %s", url)
return None

gtid = parts[-1]
ord_id = "/".join(parts[mode_idx + 1 : -1])
system_type = ord_id.split(":")[0]

if not gtid or not ord_id:
logger.debug("Skipping fragment with empty gtid or ord_id in URL: %s", url)
return None

return {
"global_tenant_id": gtid,
"system_type": system_type,
"integration_dependency": ord_id,
}
33 changes: 33 additions & 0 deletions src/sap_cloud_sdk/agentgateway/agw_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
)
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.core.telemetry import Module, Operation, record_metrics

logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -498,6 +499,38 @@ 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[dict]:
"""List all active backend system integrations for the current tenant.

Returns the connected backend systems (e.g. SAP PCE, SAP S/4HANA) that
were wired up via UCL Formations and are currently in READY state. Use
this to determine which systems are connected and which GTIDs to pass
when loading MCP tools.

Only available for LoB agents (requires tenant_subdomain 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,
Expand Down
1 change: 1 addition & 0 deletions src/sap_cloud_sdk/core/telemetry/operation.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
Loading