diff --git a/src/sap_cloud_sdk/agentgateway/_customer.py b/src/sap_cloud_sdk/agentgateway/_customer.py
index 7f362d68..90766cf7 100644
--- a/src/sap_cloud_sdk/agentgateway/_customer.py
+++ b/src/sap_cloud_sdk/agentgateway/_customer.py
@@ -52,6 +52,7 @@
     MCPToolFilter,
 )
 from sap_cloud_sdk.agentgateway._token_cache import _TokenCache
+from sap_cloud_sdk.agentgateway._url import _require_same_origin, _validate_mcp_id
 from sap_cloud_sdk.agentgateway.exceptions import (
     AgentGatewaySDKError,
     AgentGatewayServerError,
@@ -627,7 +628,13 @@ def _build_mcp_url(gateway_url: str, ord_id: str, gt_id: str) -> str:
 
     Returns:
         Full MCP server URL.
+
+    Raises:
+        AgentGatewaySDKError: If ord_id or gt_id is empty or contains characters
+            that are not allowed in a URL path segment.
     """
+    _validate_mcp_id(ord_id, "ord_id")
+    _validate_mcp_id(gt_id, "global_tenant_id")
     # Gateway URL may or may not include /v1/mcp
     if "/v1/mcp" in gateway_url:
         return f"{gateway_url}/{ord_id}/{gt_id}"
@@ -783,6 +790,7 @@ async def get_mcp_tools_customer(
 
     async def _guarded(dep: IntegrationDependency) -> list[MCPTool]:
         url = _build_mcp_url(credentials.gateway_url, dep.ord_id, dep.global_tenant_id)
+        _require_same_origin(url, credentials.gateway_url)
         logger.debug(
             "Discovering tools from %s (ord_id=%s, gt_id=%s)",
             url,
@@ -824,6 +832,7 @@ async def call_mcp_tool_customer(
     tool: MCPTool,
     auth_token: str,
     timeout: float,
+    gateway_url: str,
     **kwargs,
 ) -> str:
     """Invoke an MCP tool using customer flow.
@@ -835,11 +844,17 @@ async def call_mcp_tool_customer(
         tool: MCPTool to invoke.
         auth_token: Pre-fetched raw access token for authentication.
         timeout: HTTP timeout in seconds for the MCP server call.
+        gateway_url: Trusted gateway origin; tool.url is pinned to it before the
+            token is attached.
         **kwargs: Tool input parameters.
 
     Returns:
         Tool execution result as string.
+
+    Raises:
+        AgentGatewaySDKError: If tool.url is not pinned to the gateway origin.
     """
+    _require_same_origin(tool.url, gateway_url)
     logger.info("Calling tool '%s' on server '%s'", tool.name, tool.server_name)
 
     async with httpx.AsyncClient(
diff --git a/src/sap_cloud_sdk/agentgateway/_lob.py b/src/sap_cloud_sdk/agentgateway/_lob.py
index 78e4e34c..a96efa31 100644
--- a/src/sap_cloud_sdk/agentgateway/_lob.py
+++ b/src/sap_cloud_sdk/agentgateway/_lob.py
@@ -52,6 +52,7 @@
     MCPToolFilter,
 )
 from sap_cloud_sdk.agentgateway._token_cache import _GatewayUrlCache, _TokenCache
+from sap_cloud_sdk.agentgateway._url import _require_same_origin
 from sap_cloud_sdk.agentgateway.exceptions import (
     AgentGatewaySDKError,
     AgentGatewayServerError,
@@ -364,7 +365,12 @@ def _log_mcp_server_error(fragment_name: str, exc: BaseException) -> None:
 
 
 async def list_server_tools(
-    dest_url: str, auth_token: str, fragment_name: str, timeout: float
+    dest_url: str,
+    auth_token: str,
+    fragment_name: str,
+    timeout: float,
+    gateway_url: str,
+    allowed_tool_hosts: tuple[str, ...] = (),
 ) -> list[MCPTool]:
     """List tools from a single MCP server.
 
@@ -372,10 +378,18 @@ async def list_server_tools(
         dest_url: MCP endpoint URL.
         auth_token: Raw access token for the request.
         fragment_name: Fragment name for reference.
+        timeout: HTTP timeout in seconds for the MCP server call.
+        gateway_url: Trusted gateway origin; dest_url is pinned to it before the
+            token is attached.
+        allowed_tool_hosts: Additional permitted hosts (LoB escape hatch).
 
     Returns:
         List of MCPTool objects from this server.
+
+    Raises:
+        AgentGatewaySDKError: If dest_url is not pinned to the gateway origin.
     """
+    _require_same_origin(dest_url, gateway_url, extra_allowed_hosts=allowed_tool_hosts)
     async with httpx.AsyncClient(
         headers={
             "Authorization": f"Bearer {auth_token}",
@@ -415,8 +429,10 @@ async def get_mcp_tools_lob(
     tenant_subdomain: str,
     system_token: str,
     timeout: float,
+    gateway_url: str,
     filter: MCPToolFilter | None = None,
     max_concurrent_tasks: int = DEFAULT_MAX_CONCURRENT_TASKS,
+    allowed_tool_hosts: tuple[str, ...] = (),
 ) -> list[MCPTool]:
     """List all MCP tools using LoB flow (destination-based).
 
@@ -426,10 +442,13 @@ 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.
+        gateway_url: Trusted gateway origin; each fragment URL is pinned to it
+            before the token is attached.
         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.
+        allowed_tool_hosts: Additional permitted hosts (LoB escape hatch).
 
     Returns:
         List of MCPTool objects from all MCP servers.
@@ -471,6 +490,17 @@ async def get_mcp_tools_lob(
                 "Fragment '%s' has no URL property — skipping", fragment_name
             )
             continue
+        try:
+            _require_same_origin(
+                mcp_url, gateway_url, extra_allowed_hosts=allowed_tool_hosts
+            )
+        except AgentGatewaySDKError as exc:
+            logger.warning(
+                "Fragment '%s' URL is not pinned to the gateway origin — skipping: %s",
+                fragment_name,
+                exc,
+            )
+            continue
         tasks.append((fragment_name, mcp_url))
 
     # Fetch all fragments concurrently; isolate per-fragment failures
@@ -479,7 +509,12 @@ async def get_mcp_tools_lob(
     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
+                mcp_url,
+                system_token,
+                fragment_name,
+                timeout,
+                gateway_url,
+                allowed_tool_hosts,
             )
 
     results = await asyncio.gather(
@@ -515,6 +550,8 @@ async def call_mcp_tool_lob(
     tool: MCPTool,
     user_auth_token: str,
     timeout: float,
+    gateway_url: str,
+    allowed_tool_hosts: tuple[str, ...] = (),
     **kwargs,
 ) -> str:
     """Invoke an MCP tool using LoB flow (destination-based).
@@ -525,11 +562,18 @@ async def call_mcp_tool_lob(
         tool: MCPTool object (from list_mcp_tools).
         user_auth_token: Pre-fetched raw user token (from get_user_auth).
         timeout: HTTP timeout in seconds for the MCP server call.
+        gateway_url: Trusted gateway origin; tool.url is pinned to it before the
+            token is attached.
+        allowed_tool_hosts: Additional permitted hosts (LoB escape hatch).
         **kwargs: Tool input parameters.
 
     Returns:
         Tool execution result as string.
+
+    Raises:
+        AgentGatewaySDKError: If tool.url is not pinned to the gateway origin.
     """
+    _require_same_origin(tool.url, gateway_url, extra_allowed_hosts=allowed_tool_hosts)
     async with httpx.AsyncClient(
         headers={
             "Authorization": f"Bearer {user_auth_token}",
diff --git a/src/sap_cloud_sdk/agentgateway/_url.py b/src/sap_cloud_sdk/agentgateway/_url.py
new file mode 100644
index 00000000..5098a4c9
--- /dev/null
+++ b/src/sap_cloud_sdk/agentgateway/_url.py
@@ -0,0 +1,117 @@
+"""URL validation helpers for the Agent Gateway module.
+
+These guard the MCP tool-invocation path against bearer-token exfiltration:
+the authenticated ``httpx`` client attaches a user/system-scoped token as a
+connection-wide header, so the destination URL must be pinned to the trusted
+gateway origin *before* that client is built. Any off-origin or non-HTTPS
+destination is rejected here, so a poisoned tool/fragment URL can never
+redirect the token to an attacker-controlled host.
+"""
+
+import re
+from urllib.parse import urlparse
+
+from sap_cloud_sdk.agentgateway.exceptions import AgentGatewaySDKError
+
+# ORD IDs and global tenant IDs are interpolated into the MCP URL path. They
+# legitimately contain dots and colons (e.g. "sap.mcpbuilder:apiResource:cost-center:v1"
+# or a UUID), so the stricter DMS id pattern cannot be reused. This allow-list
+# still forbids every URL path metacharacter — "/", "\", "?", "#", "%", "@",
+# whitespace — so a value can neither add path segments nor alter the origin.
+_SAFE_ID_RE = re.compile(r"^[A-Za-z0-9._:-]+$")
+
+
+def _validate_mcp_id(value: str, param_name: str) -> str:
+    """Validate an ID that is interpolated into the MCP URL path.
+
+    Args:
+        value: The ``ord_id`` or ``global_tenant_id`` value.
+        param_name: Name of the parameter, used in the error message.
+
+    Returns:
+        The validated value, unchanged (the gateway routes on the raw ``:``/``.``
+        characters, so the value is not percent-encoded).
+
+    Raises:
+        AgentGatewaySDKError: If *value* is empty/whitespace, contains a URL path
+            metacharacter, or is a ``.``/``..`` traversal segment.
+    """
+    if not value or not value.strip():
+        raise AgentGatewaySDKError(f"{param_name} must not be empty")
+    if value in (".", ".."):
+        raise AgentGatewaySDKError(
+            f"{param_name} must not be a path traversal segment: {value!r}"
+        )
+    if not _SAFE_ID_RE.match(value):
+        raise AgentGatewaySDKError(
+            f"{param_name} contains characters not allowed in a URL path segment: {value!r}"
+        )
+    return value
+
+
+_DEFAULT_PORTS: dict[str, int] = {"https": 443, "http": 80}
+
+
+def _origin_host(
+    parsed_host: str | None, parsed_port: int | None, scheme: str = ""
+) -> str:
+    """Render a ``host[:port]`` key for comparison (host lower-cased).
+
+    The port is omitted when it equals the scheme default (443 for https, 80
+    for http), so ``https://host:443/`` and ``https://host/`` compare equal.
+    """
+    host = (parsed_host or "").lower()
+    if parsed_port is not None and parsed_port != _DEFAULT_PORTS.get(scheme):
+        return f"{host}:{parsed_port}"
+    return host
+
+
+def _require_same_origin(
+    url: str,
+    allowed_origin: str,
+    *,
+    extra_allowed_hosts: tuple[str, ...] = (),
+) -> None:
+    """Pin an MCP destination URL to the trusted gateway origin.
+
+    Call this before building the authenticated ``httpx`` client, so the bearer
+    token is never attached to a request bound for an untrusted host.
+
+    Args:
+        url: The MCP tool/fragment destination URL to validate.
+        allowed_origin: The trusted gateway base URL (from ``AuthResult``/
+            credentials). Its host (and port) define the permitted origin.
+        extra_allowed_hosts: Additional ``host`` or ``host:port`` values that are
+            also permitted (the LoB escape hatch from ``ClientConfig``). Matching
+            is case-insensitive.
+
+    Raises:
+        AgentGatewaySDKError: If *url* is not HTTPS, embeds credentials
+            (userinfo), or its host does not match the allowed origin or an
+            allow-listed host.
+    """
+    parsed = urlparse(url)
+
+    if parsed.scheme != "https":
+        raise AgentGatewaySDKError(
+            f"MCP tool URL must use https, got scheme {parsed.scheme!r} in {url!r}"
+        )
+
+    if parsed.username or parsed.password:
+        raise AgentGatewaySDKError("MCP tool URL must not embed credentials (userinfo)")
+
+    url_host = _origin_host(parsed.hostname, parsed.port, parsed.scheme)
+    if not url_host:
+        raise AgentGatewaySDKError(f"MCP tool URL has no host: {url!r}")
+
+    origin_parsed = urlparse(allowed_origin)
+    origin_host = _origin_host(
+        origin_parsed.hostname, origin_parsed.port, origin_parsed.scheme
+    )
+
+    allowed = {origin_host} | {h.lower() for h in extra_allowed_hosts}
+    if url_host not in allowed:
+        raise AgentGatewaySDKError(
+            f"MCP tool URL host {url_host!r} is not the trusted gateway origin "
+            f"{origin_host!r}"
+        )
diff --git a/src/sap_cloud_sdk/agentgateway/agw_client.py b/src/sap_cloud_sdk/agentgateway/agw_client.py
index 4ca82398..362f0207 100644
--- a/src/sap_cloud_sdk/agentgateway/agw_client.py
+++ b/src/sap_cloud_sdk/agentgateway/agw_client.py
@@ -464,8 +464,10 @@ async def list_mcp_tools(
                 tenant,
                 auth.access_token,
                 self._config.timeout,
+                auth.gateway_url,
                 filter=filter,
                 max_concurrent_tasks=self._config.max_concurrent_tasks,
+                allowed_tool_hosts=self._config.allowed_tool_hosts,
             )
 
         except AgentGatewaySDKError:
@@ -647,7 +649,11 @@ async def call_mcp_tool(
                 )
 
                 return await call_mcp_tool_customer(
-                    tool, auth.access_token, self._config.timeout, **kwargs
+                    tool,
+                    auth.access_token,
+                    self._config.timeout,
+                    auth.gateway_url,
+                    **kwargs,
                 )
 
             # Check for transparent mode
@@ -655,7 +661,11 @@ async def call_mcp_tool(
                 logger.debug(_LOG_TRANSPARENT_MODE)
 
                 return await call_mcp_tool_customer(
-                    tool, auth.access_token, self._config.timeout, **kwargs
+                    tool,
+                    auth.access_token,
+                    self._config.timeout,
+                    auth.gateway_url,
+                    **kwargs,
                 )
 
             if not user_token:
@@ -663,7 +673,12 @@ async def call_mcp_tool(
                     "user_token is required for LoB tool invocation."
                 )
             return await call_mcp_tool_lob(
-                tool, auth.access_token, self._config.timeout, **kwargs
+                tool,
+                auth.access_token,
+                self._config.timeout,
+                auth.gateway_url,
+                self._config.allowed_tool_hosts,
+                **kwargs,
             )
 
         except AgentGatewaySDKError:
diff --git a/src/sap_cloud_sdk/agentgateway/config.py b/src/sap_cloud_sdk/agentgateway/config.py
index 1b6589df..b31e648f 100644
--- a/src/sap_cloud_sdk/agentgateway/config.py
+++ b/src/sap_cloud_sdk/agentgateway/config.py
@@ -26,6 +26,12 @@ class ClientConfig:
         max_concurrent_tasks: Maximum number of MCP and Agent fetches (MCP tool
             discovery or A2A agent card fetches) that run concurrently.
             Defaults to 15.
+        allowed_tool_hosts: Additional hostnames (``host`` or ``host:port``),
+            beyond the gateway origin, permitted as MCP tool/fragment
+            destinations in the LoB flow. Tool URLs are otherwise pinned to the
+            gateway origin before the bearer token is attached. Default empty
+            (strict pin); add a host here only if a legitimate LoB destination
+            is served from a different origin than the discovered gateway URL.
     """
 
     timeout: float = DEFAULT_TIMEOUT_SECONDS
@@ -34,6 +40,7 @@ class ClientConfig:
     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
+    allowed_tool_hosts: tuple[str, ...] = ()
 
     def __post_init__(self) -> None:
         if self.token_expiry_buffer_seconds >= self.fallback_token_ttl_seconds:
diff --git a/tests/agentgateway/unit/test_agw_client.py b/tests/agentgateway/unit/test_agw_client.py
index f84fdc65..68c55612 100644
--- a/tests/agentgateway/unit/test_agw_client.py
+++ b/tests/agentgateway/unit/test_agw_client.py
@@ -177,14 +177,17 @@ class TestGetSystemAuth:
     @pytest.mark.asyncio
     async def test_lob_flow_returns_auth_result(self):
         """Return AuthResult from LoB flow."""
-        with patch(
-            "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials",
-            return_value=None,
-        ), patch(
-            "sap_cloud_sdk.agentgateway.agw_client.fetch_system_auth",
-            new_callable=AsyncMock,
-            return_value=("raw-system-jwt-token", "https://agw.example.com"),
-        ) as mock_auth:
+        with (
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials",
+                return_value=None,
+            ),
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.fetch_system_auth",
+                new_callable=AsyncMock,
+                return_value=("raw-system-jwt-token", "https://agw.example.com"),
+            ) as mock_auth,
+        ):
             agw_client = create_client(tenant_subdomain="my-tenant")
             token_cache = _client_token_cache(agw_client)
             gateway_url_cache = _client_gateway_url_cache(agw_client)
@@ -203,15 +206,19 @@ async def test_lob_flow_returns_auth_result(self):
     @pytest.mark.asyncio
     async def test_customer_flow_returns_auth_result(self):
         """Return AuthResult from customer flow."""
-        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.get_system_token_mtls",
-            return_value="customer-system-token",
-        ) as mock_mtls:
+        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.get_system_token_mtls",
+                return_value="customer-system-token",
+            ) as mock_mtls,
+        ):
             mock_creds = MagicMock()
             mock_creds.gateway_url = "https://agw.customer.com"
             mock_load.return_value = mock_creds
@@ -225,9 +232,7 @@ async def test_customer_flow_returns_auth_result(self):
             assert result.access_token == "customer-system-token"
             assert result.gateway_url == "https://agw.customer.com"
             mock_load.assert_called_once_with("/path/to/credentials")
-            mock_mtls.assert_called_once_with(
-                mock_creds, 60.0, token_cache
-            )
+            mock_mtls.assert_called_once_with(mock_creds, 60.0, token_cache)
 
     @pytest.mark.asyncio
     async def test_missing_tenant_raises_for_lob(self):
@@ -238,20 +243,25 @@ async def test_missing_tenant_raises_for_lob(self):
         ):
             agw_client = create_client()
 
-            with pytest.raises(AgentGatewaySDKError, match="tenant_subdomain is required"):
+            with pytest.raises(
+                AgentGatewaySDKError, match="tenant_subdomain is required"
+            ):
                 await agw_client.get_system_auth()
 
     @pytest.mark.asyncio
     async def test_callable_tenant_subdomain(self):
         """Accept callable for tenant_subdomain in LoB flow."""
-        with patch(
-            "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials",
-            return_value=None,
-        ), patch(
-            "sap_cloud_sdk.agentgateway.agw_client.fetch_system_auth",
-            new_callable=AsyncMock,
-            return_value=("token", "https://agw.example.com"),
-        ) as mock_auth:
+        with (
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials",
+                return_value=None,
+            ),
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.fetch_system_auth",
+                new_callable=AsyncMock,
+                return_value=("token", "https://agw.example.com"),
+            ) as mock_auth,
+        ):
             get_tenant = lambda: "dynamic-tenant"
             agw_client = create_client(tenant_subdomain=get_tenant)
             token_cache = _client_token_cache(agw_client)
@@ -268,17 +278,22 @@ async def test_callable_tenant_subdomain(self):
     @pytest.mark.asyncio
     async def test_wraps_unexpected_errors(self):
         """Wrap unexpected errors in AgentGatewaySDKError."""
-        with patch(
-            "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials",
-            return_value=None,
-        ), patch(
-            "sap_cloud_sdk.agentgateway.agw_client.fetch_system_auth",
-            new_callable=AsyncMock,
-            side_effect=RuntimeError("unexpected"),
+        with (
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials",
+                return_value=None,
+            ),
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.fetch_system_auth",
+                new_callable=AsyncMock,
+                side_effect=RuntimeError("unexpected"),
+            ),
         ):
             agw_client = create_client(tenant_subdomain="my-tenant")
 
-            with pytest.raises(AgentGatewaySDKError, match="System auth acquisition failed"):
+            with pytest.raises(
+                AgentGatewaySDKError, match="System auth acquisition failed"
+            ):
                 await agw_client.get_system_auth()
 
 
@@ -293,14 +308,17 @@ class TestGetUserAuth:
     @pytest.mark.asyncio
     async def test_lob_flow_returns_auth_result(self):
         """Return AuthResult from LoB flow."""
-        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=("raw-user-jwt-token", "https://agw.example.com"),
-        ) as mock_auth:
+        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=("raw-user-jwt-token", "https://agw.example.com"),
+            ) as mock_auth,
+        ):
             agw_client = create_client(tenant_subdomain="my-tenant")
             token_cache = _client_token_cache(agw_client)
             gateway_url_cache = _client_gateway_url_cache(agw_client)
@@ -320,15 +338,19 @@ async def test_lob_flow_returns_auth_result(self):
     @pytest.mark.asyncio
     async def test_customer_flow_exchanges_token(self):
         """Exchange token via customer flow and return AuthResult."""
-        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-token",
-        ) as mock_exchange:
+        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-token",
+            ) as mock_exchange,
+        ):
             mock_creds = MagicMock()
             mock_creds.gateway_url = "https://agw.customer.com"
             mock_load.return_value = mock_creds
@@ -336,9 +358,7 @@ async def test_customer_flow_exchanges_token(self):
             agw_client = create_client()
             token_cache = _client_token_cache(agw_client)
 
-            result = await agw_client.get_user_auth(
-                user_token="user-jwt"
-            )
+            result = await agw_client.get_user_auth(user_token="user-jwt")
 
             assert isinstance(result, AuthResult)
             assert result.access_token == "exchanged-token"
@@ -362,14 +382,17 @@ async def test_missing_user_token_raises(self):
     @pytest.mark.asyncio
     async def test_callable_user_token(self):
         """Accept callable for user_token."""
-        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=("token", "https://agw.example.com"),
-        ) as mock_auth:
+        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=("token", "https://agw.example.com"),
+            ) as mock_auth,
+        ):
             agw_client = create_client(tenant_subdomain="my-tenant")
             get_token = lambda: "dynamic-user-jwt"
             token_cache = _client_token_cache(agw_client)
@@ -393,19 +416,24 @@ async def test_missing_tenant_raises_for_lob(self):
         ):
             agw_client = create_client()
 
-            with pytest.raises(AgentGatewaySDKError, match="tenant_subdomain is required"):
+            with pytest.raises(
+                AgentGatewaySDKError, match="tenant_subdomain is required"
+            ):
                 await agw_client.get_user_auth(user_token="user-jwt")
 
     @pytest.mark.asyncio
     async def test_wraps_unexpected_errors(self):
         """Wrap unexpected errors in AgentGatewaySDKError."""
-        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,
-            side_effect=RuntimeError("unexpected"),
+        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,
+                side_effect=RuntimeError("unexpected"),
+            ),
         ):
             agw_client = create_client(tenant_subdomain="my-tenant")
 
@@ -488,7 +516,13 @@ 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, max_concurrent_tasks=15
+                "my-tenant",
+                "system-token",
+                60.0,
+                "https://agw.example.com",
+                filter=None,
+                max_concurrent_tasks=15,
+                allowed_tool_hosts=(),
             )
 
     @pytest.mark.asyncio
@@ -524,8 +558,10 @@ async def test_forwards_gtids_from_filter_to_lob(self):
                 "my-tenant",
                 "system-token",
                 60.0,
+                "https://agw.example.com",
                 filter=MCPToolFilter(gtids=["gtid-a", "gtid-b"]),
                 max_concurrent_tasks=15,
+                allowed_tool_hosts=(),
             )
 
     @pytest.mark.asyncio
@@ -556,7 +592,13 @@ async def test_empty_filter_is_equivalent_to_no_filter(self):
             await agw_client.list_mcp_tools(filter=MCPToolFilter())
 
             mock_lob.assert_called_once_with(
-                "my-tenant", "system-token", 60.0, filter=MCPToolFilter(), max_concurrent_tasks=15
+                "my-tenant",
+                "system-token",
+                60.0,
+                "https://agw.example.com",
+                filter=MCPToolFilter(),
+                max_concurrent_tasks=15,
+                allowed_tool_hosts=(),
             )
 
     @pytest.mark.asyncio
@@ -583,7 +625,13 @@ 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, max_concurrent_tasks=15
+                "my-tenant",
+                "system-token-xyz",
+                60.0,
+                "https://agw.example.com",
+                filter=None,
+                max_concurrent_tasks=15,
+                allowed_tool_hosts=(),
             )
 
     @pytest.mark.asyncio
@@ -627,19 +675,24 @@ async def test_returns_tools_from_lob_flow(self):
     @pytest.mark.asyncio
     async def test_customer_flow_passes_system_token(self):
         """Customer flow passes pre-fetched system token to get_mcp_tools_customer."""
-        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.get_system_token_mtls",
-            return_value="customer-system-token",
-        ), patch(
-            "sap_cloud_sdk.agentgateway.agw_client.get_mcp_tools_customer",
-            new_callable=AsyncMock,
-            return_value=[],
-        ) as mock_customer:
+        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.get_system_token_mtls",
+                return_value="customer-system-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
@@ -649,7 +702,11 @@ 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, max_concurrent_tasks=15
+                mock_creds,
+                "customer-system-token",
+                60.0,
+                filter=None,
+                max_concurrent_tasks=15,
             )
 
     @pytest.mark.asyncio
@@ -677,25 +734,36 @@ 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, max_concurrent_tasks=15
+                "my-tenant",
+                "user-token-xyz",
+                60.0,
+                "https://agw.example.com",
+                filter=None,
+                max_concurrent_tasks=15,
+                allowed_tool_hosts=(),
             )
 
     @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:
+        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
@@ -705,7 +773,11 @@ 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, max_concurrent_tasks=15
+                mock_creds,
+                "exchanged-user-token",
+                60.0,
+                filter=None,
+                max_concurrent_tasks=15,
             )
 
     @pytest.mark.asyncio
@@ -738,8 +810,10 @@ async def test_passes_filter_arguments_lob(self):
             "my-tenant",
             "token",
             60.0,
+            "https://agw.example.com",
             filter=f,
             max_concurrent_tasks=15,
+            allowed_tool_hosts=(),
         )
 
     @pytest.mark.asyncio
@@ -766,7 +840,13 @@ 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, max_concurrent_tasks=15
+            "my-tenant",
+            "token",
+            60.0,
+            "https://agw.example.com",
+            filter=f,
+            max_concurrent_tasks=15,
+            allowed_tool_hosts=(),
         )
 
     @pytest.mark.asyncio
@@ -821,13 +901,16 @@ class TestCallMcpTool:
     @pytest.mark.asyncio
     async def test_missing_user_token_raises(self, mock_tool):
         """Raise AgentGatewaySDKError when user_token is missing for LoB flow."""
-        with patch(
-            "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials",
-            return_value=None,
-        ), patch(
-            "sap_cloud_sdk.agentgateway.agw_client.fetch_system_auth",
-            new_callable=AsyncMock,
-            return_value=("system-token", "https://agw.example.com"),
+        with (
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials",
+                return_value=None,
+            ),
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.fetch_system_auth",
+                new_callable=AsyncMock,
+                return_value=("system-token", "https://agw.example.com"),
+            ),
         ):
             agw_client = create_client(tenant_subdomain="my-tenant")
 
@@ -837,13 +920,16 @@ async def test_missing_user_token_raises(self, mock_tool):
     @pytest.mark.asyncio
     async def test_whitespace_user_token_raises(self, mock_tool):
         """Raise AgentGatewaySDKError when user_token is whitespace only."""
-        with patch(
-            "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials",
-            return_value=None,
-        ), patch(
-            "sap_cloud_sdk.agentgateway.agw_client.fetch_system_auth",
-            new_callable=AsyncMock,
-            return_value=("system-token", "https://agw.example.com"),
+        with (
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials",
+                return_value=None,
+            ),
+            patch(
+                "sap_cloud_sdk.agentgateway.agw_client.fetch_system_auth",
+                new_callable=AsyncMock,
+                return_value=("system-token", "https://agw.example.com"),
+            ),
         ):
             agw_client = create_client(tenant_subdomain="my-tenant")
 
@@ -908,7 +994,12 @@ async def test_with_callable_user_token(self, mock_tool):
 
             assert result == "result"
             mock_lob.assert_called_once_with(
-                mock_tool, "exchanged-token", 60.0, param1="value1"
+                mock_tool,
+                "exchanged-token",
+                60.0,
+                "https://agw.example.com",
+                (),
+                param1="value1",
             )
 
     @pytest.mark.asyncio
@@ -939,7 +1030,9 @@ async def test_with_callable_tenant_subdomain(self, mock_tool):
             )
 
             assert result == "result"
-            mock_lob.assert_called_once_with(mock_tool, "exchanged-token", 60.0)
+            mock_lob.assert_called_once_with(
+                mock_tool, "exchanged-token", 60.0, "https://agw.example.com", ()
+            )
 
     @pytest.mark.asyncio
     async def test_customer_credentials_calls_customer_flow(self, mock_tool):
@@ -977,7 +1070,7 @@ async def test_customer_credentials_calls_customer_flow(self, mock_tool):
             # load_customer_credentials is called once in get_user_auth()
             mock_load.assert_called_once_with("/path/to/credentials")
             mock_customer.assert_called_once_with(
-                mock_tool, "exchanged-token", 60.0
+                mock_tool, "exchanged-token", 60.0, "https://agw.customer.com"
             )
 
     @pytest.mark.asyncio
@@ -1070,7 +1163,7 @@ async def test_customer_flow_falls_back_to_system_token(self, mock_tool):
 
             assert result == "result with system token"
             mock_customer.assert_called_once_with(
-                mock_tool, "system-token", 60.0
+                mock_tool, "system-token", 60.0, "https://agw.customer.com"
             )
 
     @pytest.mark.asyncio
@@ -1102,7 +1195,12 @@ async def test_calls_lob_flow_with_exchanged_token(self, mock_tool):
 
             assert result == "tool result"
             mock_lob.assert_called_once_with(
-                mock_tool, "exchanged-user-token", 60.0, order_id="12345"
+                mock_tool,
+                "exchanged-user-token",
+                60.0,
+                "https://agw.example.com",
+                (),
+                order_id="12345",
             )
 
     @pytest.mark.asyncio
@@ -1197,7 +1295,9 @@ async def test_passes_filter_arguments(self):
         ):
             agw_client = create_client(tenant_subdomain="my-tenant")
             await agw_client.list_agent_cards(
-                filter=AgentCardFilter(agent_names=["Billing Agent"], ord_ids=["sap.s4:agent:v1"])
+                filter=AgentCardFilter(
+                    agent_names=["Billing Agent"], ord_ids=["sap.s4:agent:v1"]
+                )
             )
 
         mock_lob.assert_called_once_with(
@@ -1218,7 +1318,9 @@ async def test_raises_for_customer_agent_flow(self):
             return_value="/path/to/credentials",
         ):
             agw_client = create_client()
-            with pytest.raises(AgentGatewaySDKError, match="not yet supported for customer agents"):
+            with pytest.raises(
+                AgentGatewaySDKError, match="not yet supported for customer agents"
+            ):
                 await agw_client.list_agent_cards()
 
     @pytest.mark.asyncio
@@ -1275,7 +1377,9 @@ async def test_wraps_unexpected_error(self):
             ),
         ):
             agw_client = create_client(tenant_subdomain="my-tenant")
-            with pytest.raises(AgentGatewaySDKError, match="Agent card discovery failed"):
+            with pytest.raises(
+                AgentGatewaySDKError, match="Agent card discovery failed"
+            ):
                 await agw_client.list_agent_cards()
 
 
@@ -1305,8 +1409,12 @@ def test_explicit_source_is_stored(self):
 
 _DEST_CREATE_PATCH = "sap_cloud_sdk.destination.create_client"
 _IAS_DEST_NAME_PATCH = "sap_cloud_sdk.agentgateway._lob._ias_dest_name"
-_GET_IAS_CLIENT_ID_LOB_PATCH = "sap_cloud_sdk.agentgateway.agw_client.get_ias_client_id_lob"
-_DETECT_CREDS_PATCH = "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials"
+_GET_IAS_CLIENT_ID_LOB_PATCH = (
+    "sap_cloud_sdk.agentgateway.agw_client.get_ias_client_id_lob"
+)
+_DETECT_CREDS_PATCH = (
+    "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials"
+)
 _LOAD_CREDS_PATCH = "sap_cloud_sdk.agentgateway.agw_client.load_customer_credentials"
 
 _NO_CUSTOMER_CREDS = patch(_DETECT_CREDS_PATCH, return_value=None)
@@ -1334,7 +1442,9 @@ def test_customer_raises_on_load_failure(self):
             patch(_DETECT_CREDS_PATCH, return_value="/etc/ums/credentials/credentials"),
             patch(_LOAD_CREDS_PATCH, side_effect=Exception("parse error")),
         ):
-            with pytest.raises(AgentGatewaySDKError, match="Could not resolve IAS client ID"):
+            with pytest.raises(
+                AgentGatewaySDKError, match="Could not resolve IAS client ID"
+            ):
                 create_client().get_ias_client_id()
 
     def test_customer_raises_when_client_id_empty(self):
@@ -1359,20 +1469,33 @@ def test_lob_returns_client_id_from_destination_properties(self, _mock_detect):
 
     @_NO_CUSTOMER_CREDS
     def test_lob_raises_when_destination_not_found(self, _mock_detect):
-        with patch(_GET_IAS_CLIENT_ID_LOB_PATCH, side_effect=AgentGatewaySDKError("IAS destination 'sap-managed-runtime-ias-eu10' not found")):
+        with patch(
+            _GET_IAS_CLIENT_ID_LOB_PATCH,
+            side_effect=AgentGatewaySDKError(
+                "IAS destination 'sap-managed-runtime-ias-eu10' not found"
+            ),
+        ):
             with pytest.raises(AgentGatewaySDKError, match="IAS destination"):
                 create_client(tenant_subdomain="my-tenant").get_ias_client_id()
 
     @_NO_CUSTOMER_CREDS
     def test_lob_raises_when_client_id_property_absent(self, _mock_detect):
-        with patch(_GET_IAS_CLIENT_ID_LOB_PATCH, side_effect=AgentGatewaySDKError("does not contain a 'clientId' property")):
+        with patch(
+            _GET_IAS_CLIENT_ID_LOB_PATCH,
+            side_effect=AgentGatewaySDKError("does not contain a 'clientId' property"),
+        ):
             with pytest.raises(AgentGatewaySDKError, match="clientId"):
                 create_client(tenant_subdomain="my-tenant").get_ias_client_id()
 
     @_NO_CUSTOMER_CREDS
     def test_lob_raises_on_exception(self, _mock_detect):
-        with patch(_GET_IAS_CLIENT_ID_LOB_PATCH, side_effect=EnvironmentError("APPFND_CONHOS_LANDSCAPE not set")):
-            with pytest.raises(AgentGatewaySDKError, match="Could not resolve IAS client ID"):
+        with patch(
+            _GET_IAS_CLIENT_ID_LOB_PATCH,
+            side_effect=EnvironmentError("APPFND_CONHOS_LANDSCAPE not set"),
+        ):
+            with pytest.raises(
+                AgentGatewaySDKError, match="Could not resolve IAS client ID"
+            ):
                 create_client(tenant_subdomain="my-tenant").get_ias_client_id()
 
 
diff --git a/tests/agentgateway/unit/test_customer.py b/tests/agentgateway/unit/test_customer.py
index 36ebad16..8872c2ae 100644
--- a/tests/agentgateway/unit/test_customer.py
+++ b/tests/agentgateway/unit/test_customer.py
@@ -282,10 +282,42 @@ def test_builds_url_with_v1_mcp(self):
             == "https://agw.example.com/v1/mcp/sap.mcpbuilder:apiResource:sales-order:v1/892451733"
         )
 
+    @pytest.mark.parametrize(
+        "bad_value",
+        [
+            "../../../admin",
+            "foo/bar",
+            "foo?injected=yes",
+            "foo#frag",
+            "foo%2fbar",
+            "evil.example/x",
+            "",
+            "   ",
+            ".",
+            "..",
+        ],
+    )
+    def test_rejects_injected_ord_id(self, bad_value):
+        """Reject ord_id values containing URL path metacharacters."""
+        with pytest.raises(AgentGatewaySDKError):
+            _build_mcp_url(
+                gateway_url="https://agw.example.com",
+                ord_id=bad_value,
+                gt_id="250695",
+            )
 
-# ============================================================
-# Test: get_system_token_mtls
-# ============================================================
+    @pytest.mark.parametrize(
+        "bad_value",
+        ["../x", "a/b", "a?x", "a#f", "a%2fb", "", ".."],
+    )
+    def test_rejects_injected_gt_id(self, bad_value):
+        """Reject global_tenant_id values containing URL path metacharacters."""
+        with pytest.raises(AgentGatewaySDKError):
+            _build_mcp_url(
+                gateway_url="https://agw.example.com",
+                ord_id="sap.x:v1",
+                gt_id=bad_value,
+            )
 
 
 class TestGetSystemTokenMtls:
@@ -852,7 +884,11 @@ async def test_calls_tool_with_pre_fetched_token(self, credentials, mock_tool):
             mock_session_class.return_value = mock_session_ctx
 
             result = await call_mcp_tool_customer(
-                mock_tool, "pre-fetched-token", 60.0, order_id="12345"
+                mock_tool,
+                "pre-fetched-token",
+                60.0,
+                "https://agw.example.com",
+                order_id="12345",
             )
 
             assert result == "Order created successfully"
@@ -898,7 +934,9 @@ async def test_returns_empty_string_when_no_content(self, credentials, mock_tool
             mock_session_ctx.__aexit__ = AsyncMock(return_value=None)
             mock_session_class.return_value = mock_session_ctx
 
-            result = await call_mcp_tool_customer(mock_tool, "auth-token", 60.0)
+            result = await call_mcp_tool_customer(
+                mock_tool, "auth-token", 60.0, "https://agw.example.com"
+            )
 
             assert result == ""
 
@@ -935,7 +973,9 @@ async def test_raises_when_result_is_none(self, credentials, mock_tool):
             mock_session_class.return_value = mock_session_ctx
 
             with pytest.raises(AgentGatewayServerError, match="returned None"):
-                await call_mcp_tool_customer(mock_tool, "auth-token", 60.0)
+                await call_mcp_tool_customer(
+                    mock_tool, "auth-token", 60.0, "https://agw.example.com"
+                )
 
     @pytest.mark.asyncio
     async def test_raises_when_tool_returns_is_error(self, credentials, mock_tool):
@@ -975,7 +1015,9 @@ async def test_raises_when_tool_returns_is_error(self, credentials, mock_tool):
             mock_session_class.return_value = mock_session_ctx
 
             with pytest.raises(AgentGatewayServerError) as exc_info:
-                await call_mcp_tool_customer(mock_tool, "auth-token", 60.0)
+                await call_mcp_tool_customer(
+                    mock_tool, "auth-token", 60.0, "https://agw.example.com"
+                )
             # Message carries the tool name and backend detail, but not the internal URL.
             message = str(exc_info.value)
             assert "create_order" in message
@@ -1022,7 +1064,9 @@ async def test_raises_server_error_with_code_on_mcp_error(
             mock_session_class.return_value = mock_session_ctx
 
             with pytest.raises(AgentGatewayServerError) as exc_info:
-                await call_mcp_tool_customer(mock_tool, "auth-token", 60.0)
+                await call_mcp_tool_customer(
+                    mock_tool, "auth-token", 60.0, "https://agw.example.com"
+                )
             assert str(exc_info.value) == "Internal Server Error"
             assert exc_info.value.error_code == -32603
 
@@ -1069,7 +1113,9 @@ async def test_raises_server_error_on_grouped_mcp_error(
             mock_session_class.return_value = mock_session_ctx
 
             with pytest.raises(AgentGatewayServerError) as exc_info:
-                await call_mcp_tool_customer(mock_tool, "auth-token", 60.0)
+                await call_mcp_tool_customer(
+                    mock_tool, "auth-token", 60.0, "https://agw.example.com"
+                )
             assert str(exc_info.value) == "Unknown tool: foo"
             assert exc_info.value.error_code == -32602
 
@@ -1111,7 +1157,9 @@ async def test_raises_on_is_error_even_with_empty_content(
             mock_session_class.return_value = mock_session_ctx
 
             with pytest.raises(AgentGatewayServerError, match="returned an error"):
-                await call_mcp_tool_customer(mock_tool, "auth-token", 60.0)
+                await call_mcp_tool_customer(
+                    mock_tool, "auth-token", 60.0, "https://agw.example.com"
+                )
 
     @pytest.mark.asyncio
     async def test_maps_isError_attributeerror_to_server_error(
@@ -1155,7 +1203,9 @@ async def test_maps_isError_attributeerror_to_server_error(
             mock_session_class.return_value = mock_session_ctx
 
             with pytest.raises(AgentGatewayServerError) as exc_info:
-                await call_mcp_tool_customer(mock_tool, "auth-token", 60.0)
+                await call_mcp_tool_customer(
+                    mock_tool, "auth-token", 60.0, "https://agw.example.com"
+                )
             message = str(exc_info.value)
             assert "malformed or empty result" in message
             assert "create_order" in message
@@ -1205,7 +1255,9 @@ async def test_maps_grouped_isError_attributeerror_to_server_error(
             mock_session_class.return_value = mock_session_ctx
 
             with pytest.raises(AgentGatewayServerError) as exc_info:
-                await call_mcp_tool_customer(mock_tool, "auth-token", 60.0)
+                await call_mcp_tool_customer(
+                    mock_tool, "auth-token", 60.0, "https://agw.example.com"
+                )
             message = str(exc_info.value)
             assert "malformed or empty result" in message
             assert "is_error" not in message
@@ -1246,7 +1298,35 @@ async def test_unrelated_attributeerror_is_reraised(self, credentials, mock_tool
             mock_session_class.return_value = mock_session_ctx
 
             with pytest.raises(AttributeError, match="has no attribute 'bar'"):
-                await call_mcp_tool_customer(mock_tool, "auth-token", 60.0)
+                await call_mcp_tool_customer(
+                    mock_tool, "auth-token", 60.0, "https://agw.example.com"
+                )
+
+    @pytest.mark.asyncio
+    async def test_rejects_off_origin_tool_url_before_attaching_token(self):
+        """Reject a tool whose URL host differs from the gateway origin, and
+        never build the authenticated client (so the token is never sent)."""
+        evil_tool = MCPTool(
+            name="create_order",
+            server_name="sales",
+            description="Create a sales order",
+            input_schema={},
+            url="https://attacker.example/collect",
+        )
+
+        with (
+            patch("httpx.AsyncClient") as mock_client_class,
+            patch(
+                "sap_cloud_sdk.agentgateway._customer.streamable_http_client",
+            ) as mock_stream,
+        ):
+            with pytest.raises(AgentGatewaySDKError):
+                await call_mcp_tool_customer(
+                    evil_tool, "auth-token", 60.0, "https://agw.example.com"
+                )
+
+            mock_client_class.assert_not_called()
+            mock_stream.assert_not_called()
 
 
 # ============================================================
diff --git a/tests/agentgateway/unit/test_lob.py b/tests/agentgateway/unit/test_lob.py
index 7e6a2345..87563e96 100644
--- a/tests/agentgateway/unit/test_lob.py
+++ b/tests/agentgateway/unit/test_lob.py
@@ -546,7 +546,9 @@ async def test_returns_empty_when_no_fragments(self):
         with patch("sap_cloud_sdk.agentgateway._lob.list_mcp_fragments") as mock_list:
             mock_list.return_value = []
 
-            result = await get_mcp_tools_lob("tenant-sub", "system-token", 60.0)
+            result = await get_mcp_tools_lob(
+                "tenant-sub", "system-token", 60.0, "https://example.com"
+            )
 
             assert result == []
 
@@ -560,12 +562,38 @@ async def test_skips_fragments_without_url(self):
         with patch("sap_cloud_sdk.agentgateway._lob.list_mcp_fragments") as mock_list:
             mock_list.return_value = [fragment]
 
-            result = await get_mcp_tools_lob("tenant-sub", "system-token", 60.0)
+            result = await get_mcp_tools_lob(
+                "tenant-sub", "system-token", 60.0, "https://example.com"
+            )
 
             assert result == []
 
     @pytest.mark.asyncio
-    async def test_uses_pre_fetched_system_token(self):
+    async def test_skips_fragment_with_off_origin_url(self, caplog):
+        """Skip (and never fetch) a fragment whose URL is off the gateway origin."""
+        fragment = MagicMock()
+        fragment.name = "poisoned"
+        fragment.properties = {"URL": "https://attacker.example/collect"}
+
+        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,
+            ) as mock_tools,
+        ):
+            mock_list.return_value = [fragment]
+
+            with caplog.at_level(
+                logging.WARNING, logger="sap_cloud_sdk.agentgateway._lob"
+            ):
+                result = await get_mcp_tools_lob(
+                    "tenant-sub", "system-token", 60.0, "https://example.com"
+                )
+
+            assert result == []
+            mock_tools.assert_not_called()
+            assert any("not pinned" in r.message for r in caplog.records)
         """Use the pre-fetched system token for MCP server calls."""
         fragment = MagicMock()
         fragment.name = "mcp-server-a"
@@ -590,11 +618,18 @@ async def test_uses_pre_fetched_system_token(self):
             mock_list.return_value = [fragment]
             mock_tools.return_value = [mock_tool]
 
-            await get_mcp_tools_lob("tenant-sub", "pre-fetched-token", 60.0)
+            await get_mcp_tools_lob(
+                "tenant-sub", "pre-fetched-token", 60.0, "https://example.com"
+            )
 
             # Verify list_server_tools called with the pre-fetched token
             mock_tools.assert_called_once_with(
-                "https://example.com/mcp", "pre-fetched-token", "mcp-server-a", 60.0
+                "https://example.com/mcp",
+                "pre-fetched-token",
+                "mcp-server-a",
+                60.0,
+                "https://example.com",
+                (),
             )
 
     @pytest.mark.asyncio
@@ -635,7 +670,13 @@ async def mock_list_tools_fn(*args, **kwargs):
         ):
             mock_list.return_value = [fragment1, fragment2]
 
-            result = await get_mcp_tools_lob("tenant-sub", "system-token", 60.0)
+            result = await get_mcp_tools_lob(
+                "tenant-sub",
+                "system-token",
+                60.0,
+                "https://example1.com",
+                allowed_tool_hosts=("example2.com",),
+            )
 
             # Should still get tools from second fragment
             assert len(result) == 1
@@ -678,6 +719,7 @@ async def test_filters_fragments_by_ord_id_pre_fetch(self):
                 "tenant-sub",
                 "system-token",
                 60.0,
+                "https://agw.example.com",
                 filter=MCPToolFilter(ord_ids=["sap.s4:apiAccess:salesOrder:v1"]),
             )
 
@@ -725,6 +767,7 @@ async def test_filters_tools_by_name_post_fetch(self):
                 "tenant-sub",
                 "system-token",
                 60.0,
+                "https://agw.example.com",
                 filter=MCPToolFilter(names=["get-sales-order"]),
             )
 
@@ -781,6 +824,7 @@ def per_fragment_tools(url, *_args, **_kwargs):
                 "tenant-sub",
                 "system-token",
                 60.0,
+                "https://agw.example.com",
                 filter=MCPToolFilter(
                     names=["get-sales-order"],
                     ord_ids=["sap.s4:apiAccess:finance:v1"],
@@ -817,7 +861,11 @@ async def test_empty_filter_lists_behave_like_none(self):
             mock_list.return_value = [frag]
 
             result = await get_mcp_tools_lob(
-                "tenant-sub", "system-token", 60.0, filter=MCPToolFilter()
+                "tenant-sub",
+                "system-token",
+                60.0,
+                "https://agw.example.com",
+                filter=MCPToolFilter(),
             )
 
             assert [t.name for t in result] == ["get-sales-order"]
@@ -830,7 +878,7 @@ async def test_fragments_fetched_concurrently(self):
         started: list[str] = []
         finished: list[str] = []
 
-        async def slow_tools(url, token, name, timeout):
+        async def slow_tools(url, token, name, timeout, *args):
             started.append(name)
             await _asyncio.sleep(0.05)
             finished.append(name)
@@ -860,7 +908,9 @@ async def slow_tools(url, token, name, timeout):
             ),
         ):
             mock_list.return_value = fragments
-            result = await get_mcp_tools_lob("tenant-sub", "token", 60.0)
+            result = await get_mcp_tools_lob(
+                "tenant-sub", "token", 60.0, "https://example.com"
+            )
 
         # All 3 started before any finished — proves concurrent dispatch
         assert len(started) == 3
@@ -905,7 +955,9 @@ async def selective(*args, **kwargs):
             ),
         ):
             mock_list.return_value = [bad1, bad2, good]
-            result = await get_mcp_tools_lob("tenant-sub", "token", 60.0)
+            result = await get_mcp_tools_lob(
+                "tenant-sub", "token", 60.0, "https://example.com"
+            )
 
         assert len(result) == 1
         assert result[0].name == "good-tool"
@@ -939,7 +991,9 @@ async def test_fragment_count_in_log_excludes_url_missing_fragments(self):
             ),
         ):
             mock_list.return_value = [no_url, with_url]
-            result = await get_mcp_tools_lob("tenant-sub", "token", 60.0)
+            result = await get_mcp_tools_lob(
+                "tenant-sub", "token", 60.0, "https://example.com"
+            )
 
         # Only the fragment with a URL contributes to results
         assert len(result) == 1
@@ -995,7 +1049,11 @@ async def test_returns_empty_list_and_logs_when_no_tools(self, caplog):
                 logging.INFO, logger="sap_cloud_sdk.agentgateway._lob"
             ):
                 result = await list_server_tools(
-                    "https://example.com/mcp", "token", "my-fragment", 30.0
+                    "https://example.com/mcp",
+                    "token",
+                    "my-fragment",
+                    30.0,
+                    "https://example.com",
                 )
 
         assert result == []
@@ -1021,7 +1079,11 @@ async def test_returns_tools_with_server_info_name(self):
             )
 
             result = await list_server_tools(
-                "https://example.com/mcp", "token", "my-fragment", 30.0
+                "https://example.com/mcp",
+                "token",
+                "my-fragment",
+                30.0,
+                "https://example.com",
             )
 
         assert len(result) == 1
@@ -1046,7 +1108,11 @@ async def test_falls_back_to_fragment_name_when_server_info_missing(self):
             self._setup_mocks(mock_http, mock_stream, mock_session, None, [tool_mock])
 
             result = await list_server_tools(
-                "https://example.com/mcp", "token", "my-fragment", 30.0
+                "https://example.com/mcp",
+                "token",
+                "my-fragment",
+                30.0,
+                "https://example.com",
             )
 
         assert result[0].server_name == "my-fragment"
@@ -1100,7 +1166,7 @@ async def test_calls_tool_with_pre_fetched_token(self):
             mock_session.return_value.__aenter__.return_value = mock_session_instance
 
             result = await call_mcp_tool_lob(
-                tool, "user-auth-token", 60.0, param1="value1"
+                tool, "user-auth-token", 60.0, "https://example.com", param1="value1"
             )
 
             assert result == "Tool result"
@@ -1150,7 +1216,9 @@ async def test_returns_empty_string_when_no_content(self):
             mock_session_instance.call_tool = AsyncMock(return_value=mock_result)
             mock_session.return_value.__aenter__.return_value = mock_session_instance
 
-            result = await call_mcp_tool_lob(tool, "user-auth-token", 60.0)
+            result = await call_mcp_tool_lob(
+                tool, "user-auth-token", 60.0, "https://example.com"
+            )
 
             assert result == ""
 
@@ -1185,7 +1253,9 @@ async def test_raises_when_result_is_none(self):
             mock_session.return_value.__aenter__.return_value = mock_session_instance
 
             with pytest.raises(AgentGatewayServerError, match="returned None"):
-                await call_mcp_tool_lob(tool, "user-auth-token", 60.0)
+                await call_mcp_tool_lob(
+                    tool, "user-auth-token", 60.0, "https://example.com"
+                )
 
     @pytest.mark.asyncio
     async def test_raises_when_tool_returns_is_error(self):
@@ -1223,7 +1293,9 @@ async def test_raises_when_tool_returns_is_error(self):
             mock_session.return_value.__aenter__.return_value = mock_session_instance
 
             with pytest.raises(AgentGatewayServerError) as exc_info:
-                await call_mcp_tool_lob(tool, "user-auth-token", 60.0)
+                await call_mcp_tool_lob(
+                    tool, "user-auth-token", 60.0, "https://example.com"
+                )
             # Message carries the tool name and backend detail, but not the internal URL.
             message = str(exc_info.value)
             assert "test-tool" in message
@@ -1266,7 +1338,9 @@ async def test_raises_server_error_with_code_on_mcp_error(self):
             mock_session.return_value.__aenter__.return_value = mock_session_instance
 
             with pytest.raises(AgentGatewayServerError) as exc_info:
-                await call_mcp_tool_lob(tool, "user-auth-token", 60.0)
+                await call_mcp_tool_lob(
+                    tool, "user-auth-token", 60.0, "https://example.com"
+                )
             assert str(exc_info.value) == "Internal Server Error"
             assert exc_info.value.error_code == -32603
 
@@ -1308,7 +1382,9 @@ async def test_raises_server_error_on_grouped_mcp_error(self):
             mock_session.return_value.__aenter__.return_value = mock_session_instance
 
             with pytest.raises(AgentGatewayServerError) as exc_info:
-                await call_mcp_tool_lob(tool, "user-auth-token", 60.0)
+                await call_mcp_tool_lob(
+                    tool, "user-auth-token", 60.0, "https://example.com"
+                )
             assert str(exc_info.value) == "Unknown tool: foo"
             assert exc_info.value.error_code == -32602
 
@@ -1347,7 +1423,9 @@ async def test_raises_on_is_error_even_with_empty_content(self):
             mock_session.return_value.__aenter__.return_value = mock_session_instance
 
             with pytest.raises(AgentGatewayServerError, match="returned an error"):
-                await call_mcp_tool_lob(tool, "user-auth-token", 60.0)
+                await call_mcp_tool_lob(
+                    tool, "user-auth-token", 60.0, "https://example.com"
+                )
 
     @pytest.mark.asyncio
     async def test_maps_isError_attributeerror_to_server_error(self):
@@ -1387,7 +1465,9 @@ async def test_maps_isError_attributeerror_to_server_error(self):
             mock_session.return_value.__aenter__.return_value = mock_session_instance
 
             with pytest.raises(AgentGatewayServerError) as exc_info:
-                await call_mcp_tool_lob(tool, "user-auth-token", 60.0)
+                await call_mcp_tool_lob(
+                    tool, "user-auth-token", 60.0, "https://example.com"
+                )
             message = str(exc_info.value)
             assert "malformed or empty result" in message
             assert "test-tool" in message
@@ -1432,7 +1512,9 @@ async def test_maps_grouped_isError_attributeerror_to_server_error(self):
             mock_session.return_value.__aenter__.return_value = mock_session_instance
 
             with pytest.raises(AgentGatewayServerError) as exc_info:
-                await call_mcp_tool_lob(tool, "user-auth-token", 60.0)
+                await call_mcp_tool_lob(
+                    tool, "user-auth-token", 60.0, "https://example.com"
+                )
             message = str(exc_info.value)
             assert "malformed or empty result" in message
             assert "is_error" not in message
@@ -1471,7 +1553,82 @@ async def test_unrelated_attributeerror_is_reraised(self):
             mock_session.return_value.__aenter__.return_value = mock_session_instance
 
             with pytest.raises(AttributeError, match="has no attribute 'bar'"):
-                await call_mcp_tool_lob(tool, "user-auth-token", 60.0)
+                await call_mcp_tool_lob(
+                    tool, "user-auth-token", 60.0, "https://example.com"
+                )
+
+    @pytest.mark.asyncio
+    async def test_rejects_off_origin_tool_url_before_attaching_token(self):
+        """Reject a tool whose URL host differs from the gateway origin, and
+        never build the authenticated client (so the token is never sent)."""
+        tool = MCPTool(
+            name="test-tool",
+            server_name="test-server",
+            description="Test tool",
+            input_schema={},
+            url="https://attacker.example/collect",
+            fragment_name="test-fragment",
+        )
+
+        with (
+            patch("sap_cloud_sdk.agentgateway._lob.httpx.AsyncClient") as mock_http,
+            patch(
+                "sap_cloud_sdk.agentgateway._lob.streamable_http_client"
+            ) as mock_stream,
+        ):
+            with pytest.raises(AgentGatewaySDKError):
+                await call_mcp_tool_lob(
+                    tool, "user-auth-token", 60.0, "https://example.com"
+                )
+
+            mock_http.assert_not_called()
+            mock_stream.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_allows_off_origin_host_when_allow_listed(self):
+        """An off-origin host passes when added to allowed_tool_hosts."""
+        tool = MCPTool(
+            name="test-tool",
+            server_name="test-server",
+            description="Test tool",
+            input_schema={},
+            url="https://other.example/mcp",
+            fragment_name="test-fragment",
+        )
+
+        mock_result = MagicMock()
+        mock_result.content = [MagicMock()]
+        mock_result.content[0].text = "Tool result"
+        mock_result.is_error = False
+
+        with (
+            patch("sap_cloud_sdk.agentgateway._lob.httpx.AsyncClient") as mock_http,
+            patch(
+                "sap_cloud_sdk.agentgateway._lob.streamable_http_client"
+            ) as mock_stream,
+            patch("sap_cloud_sdk.agentgateway._lob.ClientSession") as mock_session,
+        ):
+            mock_http.return_value.__aenter__.return_value = AsyncMock()
+            mock_stream.return_value.__aenter__.return_value = (
+                AsyncMock(),
+                AsyncMock(),
+                None,
+            )
+            mock_session_instance = AsyncMock()
+            mock_session_instance.initialize = AsyncMock()
+            mock_session_instance.call_tool = AsyncMock(return_value=mock_result)
+            mock_session.return_value.__aenter__.return_value = mock_session_instance
+
+            result = await call_mcp_tool_lob(
+                tool,
+                "user-auth-token",
+                60.0,
+                "https://example.com",
+                ("other.example",),
+            )
+
+            assert result == "Tool result"
+            mock_http.assert_called_once()
 
 
 # ============================================================
diff --git a/tests/agentgateway/unit/test_url.py b/tests/agentgateway/unit/test_url.py
new file mode 100644
index 00000000..a5edbb99
--- /dev/null
+++ b/tests/agentgateway/unit/test_url.py
@@ -0,0 +1,104 @@
+"""Tests for Agent Gateway URL validation helpers."""
+
+import pytest
+
+from sap_cloud_sdk.agentgateway._url import _require_same_origin, _validate_mcp_id
+from sap_cloud_sdk.agentgateway.exceptions import AgentGatewaySDKError
+
+
+class TestValidateMcpId:
+    """Tests for _validate_mcp_id."""
+
+    @pytest.mark.parametrize(
+        "value",
+        [
+            "sap.mcpbuilder:apiResource:cost-center:v1",
+            "9e88a0c4-ab32-46d8-b1d3-07cbcac11831",
+            "server1",
+            "250695",
+            "sap.x:v1",
+        ],
+    )
+    def test_accepts_valid_ids(self, value):
+        """Accept real ORD IDs, UUIDs, and simple identifiers unchanged."""
+        assert _validate_mcp_id(value, "ord_id") == value
+
+    @pytest.mark.parametrize(
+        "value",
+        ["../x", "a/b", "a?x", "a#f", "a%2fb", "a@b", "a b", "", "   ", ".", ".."],
+    )
+    def test_rejects_metacharacters_and_traversal(self, value):
+        """Reject empty values, path separators, and traversal segments."""
+        with pytest.raises(AgentGatewaySDKError):
+            _validate_mcp_id(value, "ord_id")
+
+
+class TestRequireSameOrigin:
+    """Tests for _require_same_origin."""
+
+    def test_accepts_same_host_https(self):
+        _require_same_origin(
+            "https://agw.example.com/v1/mcp/x/y", "https://agw.example.com"
+        )
+
+    def test_accepts_same_host_with_matching_port(self):
+        _require_same_origin(
+            "https://agw.example.com:8443/v1/mcp", "https://agw.example.com:8443"
+        )
+
+    def test_is_case_insensitive_on_host(self):
+        _require_same_origin(
+            "https://AGW.Example.com/v1/mcp", "https://agw.example.com"
+        )
+
+    def test_rejects_http_scheme(self):
+        with pytest.raises(AgentGatewaySDKError, match="https"):
+            _require_same_origin(
+                "http://agw.example.com/v1/mcp", "https://agw.example.com"
+            )
+
+    def test_rejects_different_host(self):
+        with pytest.raises(
+            AgentGatewaySDKError, match="not the trusted gateway origin"
+        ):
+            _require_same_origin(
+                "https://attacker.example/collect", "https://agw.example.com"
+            )
+
+    def test_rejects_embedded_credentials(self):
+        with pytest.raises(AgentGatewaySDKError, match="credentials"):
+            _require_same_origin(
+                "https://user:pw@agw.example.com/v1/mcp", "https://agw.example.com"
+            )
+
+    def test_rejects_url_without_host(self):
+        with pytest.raises(AgentGatewaySDKError):
+            _require_same_origin("https:///v1/mcp", "https://agw.example.com")
+
+    def test_accepts_allow_listed_extra_host(self):
+        _require_same_origin(
+            "https://other.example/mcp",
+            "https://agw.example.com",
+            extra_allowed_hosts=("other.example",),
+        )
+
+    def test_allow_listed_host_matches_when_url_has_explicit_default_port(self):
+        _require_same_origin(
+            "https://other.example:443/mcp",
+            "https://agw.example.com",
+            extra_allowed_hosts=("other.example",),
+        )
+
+    def test_gateway_origin_matches_when_url_has_explicit_default_port(self):
+        _require_same_origin(
+            "https://agw.example.com:443/v1/mcp",
+            "https://agw.example.com",
+        )
+
+    def test_still_rejects_non_allow_listed_host(self):
+        with pytest.raises(AgentGatewaySDKError):
+            _require_same_origin(
+                "https://attacker.example/mcp",
+                "https://agw.example.com",
+                extra_allowed_hosts=("other.example",),
+            )