feat: support principal propagation on list mcp tools

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "sap-cloud-sdk"
-version = "0.24.0"
+version = "0.25.0"
 description = "SAP Cloud SDK for Python"
 readme = "README.md"
 license = "Apache-2.0"

src/sap_cloud_sdk/agentgateway/agw_client.py

@@ -292,6 +292,7 @@ async def get_user_auth(
     @record_metrics(Module.AGENTGATEWAY, Operation.AGENTGATEWAY_LIST_MCP_TOOLS)
     async def list_mcp_tools(
         self,
+        user_token: str | Callable[[], str] | None = None,
         app_tid: str | None = None,
     ) -> list[MCPTool]:
         """List all MCP tools from MCP servers.
@@ -301,10 +302,16 @@ async def list_mcp_tools(
 
         For LoB agents: Uses Phase 1 auth (client-scoped) via BTP Destination Service.
             Tools are auto-discovered from destination fragments.
+            If user_token is provided, uses Phase 2 auth (user-scoped) instead.
         For Customer agents: Uses mTLS client credentials.
             Tools are discovered from all servers in credentials integrationDependencies.
+            If user_token is provided, uses token exchange (jwt-bearer) instead of
+            system token.
 
         Args:
+            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.
             app_tid: BTP Application Tenant ID of the subscriber.
                 Only used for customer agents.
 
@@ -319,6 +326,9 @@ async def list_mcp_tools(
             tools = await agw_client.list_mcp_tools()
             for tool in tools:
                 print(f"{tool.name}: {tool.description}")
+
+            # With user token for principal propagation:
+            tools = await agw_client.list_mcp_tools(user_token="user-jwt")
             ```
         """
         try:
@@ -329,7 +339,10 @@ async def list_mcp_tools(
                     "Customer agent credentials detected at '%s'", credentials_path
                 )
                 credentials = load_customer_credentials(credentials_path)
-                auth = await self.get_system_auth(app_tid=app_tid)
+                if user_token:
+                    auth = await self.get_user_auth(user_token, app_tid)
+                else:
+                    auth = await self.get_system_auth(app_tid=app_tid)
                 return await get_mcp_tools_customer(
                     credentials, auth.access_token, self._config.timeout
                 )
@@ -339,7 +352,10 @@ async def list_mcp_tools(
                 logger.warning("app_tid parameter ignored for LoB agent flow")
 
             tenant = self._resolve_tenant_subdomain()
-            auth = await self.get_system_auth()
+            if user_token:
+                auth = await self.get_user_auth(user_token)
+            else:
+                auth = await self.get_system_auth()
             return await get_mcp_tools_lob(
                 tenant, auth.access_token, self._config.timeout
             )

src/sap_cloud_sdk/agentgateway/user-guide.md

@@ -29,6 +29,9 @@ tools = await agw_client.list_mcp_tools()
 for tool in tools:
     print(f"{tool.name}: {tool.description}")
 
+# Discover tools with user principal propagation
+tools = await agw_client.list_mcp_tools(user_token="user-jwt")
+
 # Invoke a tool with user principal propagation
 result = await agw_client.call_mcp_tool(
     tool=tools[0],
@@ -51,6 +54,9 @@ agw_client = create_client(tenant_subdomain="my-tenant", config=config)
 # Discover tools (auto-discovered from destination fragments)
 tools = await agw_client.list_mcp_tools()
 
+# Discover tools with user principal propagation
+tools = await agw_client.list_mcp_tools(user_token="user-jwt")
+
 # Invoke a tool (user_token required for principal propagation)
 result = await agw_client.call_mcp_tool(
     tool=tools[0],
@@ -139,6 +145,7 @@ The SDK keeps token caches per `AgentGatewayClient` instance and reuses valid ca
 class AgentGatewayClient:
     async def list_mcp_tools(
         self,
+        user_token: str | Callable[[], str] | None = None,
         app_tid: str | None = None,
     ) -> list[MCPTool]
 

tests/agentgateway/unit/test_agw_client.py

@@ -521,6 +521,60 @@ async def test_customer_flow_passes_system_token(self):
                 mock_creds, "customer-system-token", 60.0
             )
 
+    @pytest.mark.asyncio
+    async def test_lob_flow_with_user_token_uses_user_auth(self):
+        """LoB flow uses user auth when user_token is provided."""
+        with (
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials",
+                return_value=None,
+            ),
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.fetch_user_auth",
+                new_callable=AsyncMock,
+                return_value=("user-token-xyz", "https://agw.example.com"),
+            ) as mock_user_auth,
+            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(user_token="user-jwt")
+
+            mock_user_auth.assert_called_once()
+            mock_lob.assert_called_once_with("my-tenant", "user-token-xyz", 60.0)
+
+    @pytest.mark.asyncio
+    async def test_customer_flow_with_user_token_uses_user_auth(self):
+        """Customer flow uses user auth when user_token is provided."""
+        with patch(
+            "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials",
+            return_value="/path/to/credentials",
+        ), patch(
+            "sap_cloud_sdk.agentgateway.agw_client.load_customer_credentials",
+        ) as mock_load, patch(
+            "sap_cloud_sdk.agentgateway.agw_client.exchange_user_token",
+            return_value="exchanged-user-token",
+        ), patch(
+            "sap_cloud_sdk.agentgateway.agw_client.get_mcp_tools_customer",
+            new_callable=AsyncMock,
+            return_value=[],
+        ) as mock_customer:
+            mock_creds = MagicMock()
+            mock_creds.gateway_url = "https://agw.customer.com"
+            mock_load.return_value = mock_creds
+
+            agw_client = create_client()
+
+            await agw_client.list_mcp_tools(user_token="user-jwt", app_tid="tid")
+
+            mock_customer.assert_called_once_with(
+                mock_creds, "exchanged-user-token", 60.0
+            )
+
 
 # ============================================================
 # Test: call_mcp_tool