feat(mcp): graft client_credentials (M2M) to the v2 resolver

First real port body for the graft: HttpxClientCredentialsFetcher (v2_port_bodies.py), an
imperative-shell adapter that runs the RFC 6749 client_credentials grant via litellm's
configured httpx client, mirroring v1's grant for parity (client_id/client_secret/scope in the
body, parse access_token/expires_in, default TTL 3600; the arm's 60s _REFRESH_BUFFER handles
re-mint, matching v1's default buffer). The token response is validated with a pydantic model
rather than poking raw JSON. Error mapping: rejected grant (4xx) -> misconfigured (500, the
gateway's service-account config is wrong), endpoint down / 5xx / network -> upstream_unavailable
(503).

Wired into the bridge composition root (replaces the _Unwired fetcher; service-token store stays
in-memory, matching v1's per-worker caching) and grafted: _to_server_spec now maps v1 M2M
servers (oauth2 + oauth2_flow=client_credentials, i.e. has_client_credentials) to
ClientCredentialsConfig. The body lives on the v1/integration side so the v2 core keeps its
no-v1-imports invariant. The SDK's ClientCredentialsOAuthProvider is deferred to Phase 2 (it is
a connection-session-coupled httpx.Auth, attached when v2 owns the upstream MCP transport).

Tests: fetcher grant shape / token parsing / error mapping (rejected->500, 5xx->503,
network->503, missing access_token->500), the M2M adapter mapping, and an end-to-end graft test
(mocked token endpoint -> Bearer header). 74 tests pass; bridge typechecks clean.

litellm/proxy/_experimental/mcp_server/v2_port_bodies.py

@@ -0,0 +1,109 @@
+"""Real (imperative-shell) bodies for the v2 outbound-credential ports.
+
+These adapters bridge the clean-room v2 ports to real infrastructure (litellm's HTTP client, v1
+storage, ...). They live on the v1/integration side so the v2 core keeps its no-v1-imports
+invariant; the graft composition root injects them and v2 unit tests fake them.
+"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+from typing import Optional
+
+from pydantic import BaseModel, ConfigDict, SecretStr, ValidationError
+
+from litellm.constants import MCP_OAUTH2_TOKEN_CACHE_DEFAULT_TTL
+from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
+from litellm.proxy.gateway.mcp.outbound_credentials.clock import Clock, SystemClock
+from litellm.proxy.gateway.mcp.outbound_credentials.token_store import StoredToken
+from litellm.proxy.gateway.mcp.outbound_credentials.types import (
+    ClientCredentialsConfig,
+    CredError,
+)
+from litellm.proxy.gateway.mcp.result import Error, Ok, Result
+from litellm.types.llms.custom_http import httpxSpecialProvider
+
+
+class _TokenResponse(BaseModel):
+    """The slice of an OAuth2 token response we consume; extra fields are ignored."""
+
+    model_config = ConfigDict(extra="ignore")
+    access_token: str
+    expires_in: Optional[int] = None
+
+
+class HttpxClientCredentialsFetcher:
+    """RFC 6749 client_credentials grant via litellm's configured httpx client.
+
+    Phase-1 graft body, mirroring v1's grant (`oauth2_token_cache._fetch_token`) for parity:
+    `client_id` / `client_secret` / `scope` in the POST body, parse `access_token` /
+    `expires_in`. Phase 2, when v2 owns the upstream MCP transport, swaps to the SDK's
+    `ClientCredentialsOAuthProvider` (a connection-session-coupled `httpx.Auth`, not a fetcher).
+    """
+
+    def __init__(self, clock: Optional[Clock] = None) -> None:
+        self._clock: Clock = clock or SystemClock()
+
+    async def fetch(
+        self, config: ClientCredentialsConfig
+    ) -> Result[StoredToken, CredError]:
+        data = {
+            "grant_type": "client_credentials",
+            "client_id": config.client_id,
+            "client_secret": config.client_secret.get_secret_value(),
+        }
+        if config.scopes:
+            data["scope"] = " ".join(config.scopes)
+
+        client = get_async_httpx_client(llm_provider=httpxSpecialProvider.MCP)
+        try:
+            response = await client.post(config.token_url, data=data)
+        except Exception as e:  # network / timeout / DNS
+            return Error(
+                CredError.of_upstream_unavailable(
+                    f"client_credentials token endpoint unreachable: {e}"
+                )
+            )
+
+        if response is None:
+            return Error(
+                CredError.of_upstream_unavailable(
+                    "client_credentials token endpoint returned no response"
+                )
+            )
+        if response.status_code >= 500:
+            return Error(
+                CredError.of_upstream_unavailable(
+                    f"client_credentials token endpoint returned {response.status_code}"
+                )
+            )
+        if response.status_code >= 400:
+            return Error(
+                CredError.of_misconfigured(
+                    f"client_credentials grant rejected ({response.status_code})"
+                )
+            )
+
+        try:
+            parsed = _TokenResponse.model_validate(response.json())
+        except ValidationError:
+            return Error(
+                CredError.of_misconfigured(
+                    "client_credentials response missing a valid access_token"
+                )
+            )
+        # Store the raw lifetime; the resolver arm's _REFRESH_BUFFER (60s, = v1's default
+        # buffer) handles proactive re-mint, so no buffer is subtracted here.
+        ttl = timedelta(
+            seconds=(
+                parsed.expires_in
+                if parsed.expires_in is not None
+                else MCP_OAUTH2_TOKEN_CACHE_DEFAULT_TTL
+            )
+        )
+        return Ok(
+            StoredToken(
+                access_token=SecretStr(parsed.access_token),
+                expires_at=self._clock.now() + ttl,
+            )
+        )

litellm/proxy/_experimental/mcp_server/v2_resolver_bridge.py

@@ -20,6 +20,9 @@ import httpx
 from pydantic import SecretStr
 
 from litellm._logging import verbose_logger
+from litellm.proxy._experimental.mcp_server.v2_port_bodies import (
+    HttpxClientCredentialsFetcher,
+)
 from litellm.proxy.gateway.mcp.outbound_credentials.clock import SystemClock
 from litellm.proxy.gateway.mcp.outbound_credentials.credential_store import (
     InMemoryCredentialStore,
@@ -94,7 +97,7 @@ def _provider() -> UpstreamCredentialProvider:
         token_refresher=unwired,
         clock=SystemClock(),
         service_token_store=InMemoryServiceTokenStore(),
-        client_credentials_fetcher=unwired,
+        client_credentials_fetcher=HttpxClientCredentialsFetcher(),
         token_exchanger=unwired,
         signer_factory=unwired,
     )
@@ -119,6 +122,22 @@ def _to_server_spec(server: MCPServer) -> Optional[ServerSpec]:
                 key_source=SharedKey(value=SecretStr(token)),
             ),
         )
+    if (
+        server.has_client_credentials
+        and server.client_id
+        and server.client_secret
+        and server.token_url
+    ):
+        return ServerSpec(
+            server_id=server.server_id,
+            resource=resource,
+            config=ClientCredentialsConfig(
+                client_id=server.client_id,
+                client_secret=SecretStr(server.client_secret),
+                token_url=server.token_url,
+                scopes=tuple(server.scopes or ()),
+            ),
+        )
     return None  # other modes are not grafted yet
 
 

tests/test_litellm/proxy/_experimental/mcp_server/test_v2_port_bodies.py

@@ -0,0 +1,103 @@
+"""Unit tests for the v2 port bodies (imperative-shell adapters).
+
+Covers the client_credentials fetcher's grant shape, token parsing, and error mapping.
+"""
+
+import httpx
+import pytest
+from pydantic import SecretStr
+
+from litellm.proxy._experimental.mcp_server import v2_port_bodies
+from litellm.proxy._experimental.mcp_server.v2_port_bodies import (
+    HttpxClientCredentialsFetcher,
+)
+from litellm.proxy.gateway.mcp.outbound_credentials.types import ClientCredentialsConfig
+from litellm.proxy.gateway.mcp.result import Error, Ok
+
+pytestmark = pytest.mark.asyncio
+
+
+class _FakeResponse:
+    def __init__(self, status_code, json_body):
+        self.status_code = status_code
+        self._json = json_body
+
+    def json(self):
+        return self._json
+
+
+class _FakeClient:
+    def __init__(self, response=None, exc=None):
+        self._response = response
+        self._exc = exc
+        self.posted = None
+
+    async def post(self, url, data=None):
+        self.posted = {"url": url, "data": data}
+        if self._exc is not None:
+            raise self._exc
+        return self._response
+
+
+def _cfg(scopes=()):
+    return ClientCredentialsConfig(
+        client_id="cid",
+        client_secret=SecretStr("secret"),
+        token_url="https://idp/token",
+        scopes=tuple(scopes),
+    )
+
+
+def _patch(monkeypatch, response=None, exc=None):
+    fake = _FakeClient(response=response, exc=exc)
+    monkeypatch.setattr(v2_port_bodies, "get_async_httpx_client", lambda **kw: fake)
+    return fake
+
+
+async def test_fetch_success_builds_stored_token(monkeypatch):
+    fake = _patch(
+        monkeypatch, _FakeResponse(200, {"access_token": "tok-123", "expires_in": 3600})
+    )
+    result = await HttpxClientCredentialsFetcher().fetch(_cfg(scopes=["a", "b"]))
+    assert isinstance(result, Ok)
+    assert result.ok.access_token.get_secret_value() == "tok-123"
+    # grant goes in the POST body, scope is space-joined (mirrors v1)
+    assert fake.posted["url"] == "https://idp/token"
+    assert fake.posted["data"]["grant_type"] == "client_credentials"
+    assert fake.posted["data"]["client_id"] == "cid"
+    assert fake.posted["data"]["client_secret"] == "secret"
+    assert fake.posted["data"]["scope"] == "a b"
+
+
+async def test_fetch_omits_scope_when_none(monkeypatch):
+    fake = _patch(monkeypatch, _FakeResponse(200, {"access_token": "t"}))
+    await HttpxClientCredentialsFetcher().fetch(_cfg())
+    assert "scope" not in fake.posted["data"]
+
+
+async def test_fetch_rejected_grant_is_misconfigured(monkeypatch):
+    _patch(monkeypatch, _FakeResponse(400, {"error": "invalid_client"}))
+    result = await HttpxClientCredentialsFetcher().fetch(_cfg())
+    assert isinstance(result, Error)
+    assert result.error.tag == "misconfigured"
+
+
+async def test_fetch_server_error_is_upstream_unavailable(monkeypatch):
+    _patch(monkeypatch, _FakeResponse(503, {}))
+    result = await HttpxClientCredentialsFetcher().fetch(_cfg())
+    assert isinstance(result, Error)
+    assert result.error.tag == "upstream_unavailable"
+
+
+async def test_fetch_network_error_is_upstream_unavailable(monkeypatch):
+    _patch(monkeypatch, exc=httpx.ConnectError("boom"))
+    result = await HttpxClientCredentialsFetcher().fetch(_cfg())
+    assert isinstance(result, Error)
+    assert result.error.tag == "upstream_unavailable"
+
+
+async def test_fetch_missing_access_token_is_misconfigured(monkeypatch):
+    _patch(monkeypatch, _FakeResponse(200, {"token_type": "bearer"}))
+    result = await HttpxClientCredentialsFetcher().fetch(_cfg())
+    assert isinstance(result, Error)
+    assert result.error.tag == "misconfigured"

tests/test_litellm/proxy/_experimental/mcp_server/test_v2_resolver_bridge.py

@@ -85,3 +85,57 @@ async def test_resolve_mcp_auth_hook_uses_v1_when_off(v2_off):
     # v1 returns the raw token string; MCPClient then maps it to the X-API-Key header
     server = _server(MCPAuth.api_key, "up-secret")
     assert await resolve_mcp_auth(server) == "up-secret"
+
+
+def _m2m_server():
+    return MCPServer(
+        server_id="m2m",
+        name="m2m",
+        transport=MCPTransport.http,
+        url="https://up.example/mcp",
+        auth_type=MCPAuth.oauth2,
+        oauth2_flow="client_credentials",
+        client_id="cid",
+        client_secret="csecret",
+        token_url="https://idp/token",
+        scopes=["a", "b"],
+    )
+
+
+async def test_client_credentials_maps_to_config():
+    from litellm.proxy._experimental.mcp_server.v2_resolver_bridge import (
+        _to_server_spec,
+    )
+    from litellm.proxy.gateway.mcp.outbound_credentials.types import (
+        ClientCredentialsConfig,
+    )
+
+    spec = _to_server_spec(_m2m_server())
+    assert spec is not None
+    assert isinstance(spec.config, ClientCredentialsConfig)
+    assert spec.config.client_id == "cid"
+    assert spec.config.token_url == "https://idp/token"
+    assert spec.config.client_secret.get_secret_value() == "csecret"
+    assert spec.config.scopes == ("a", "b")
+
+
+async def test_client_credentials_graft_end_to_end(v2_on, monkeypatch):
+    # M2M flows through the real fetcher; mock the IdP token endpoint and assert the Bearer.
+    from litellm.proxy._experimental.mcp_server import v2_port_bodies
+
+    class _Resp:
+        status_code = 200
+
+        def json(self):
+            return {"access_token": "m2m-tok", "expires_in": 3600}
+
+    class _Client:
+        async def post(self, url, data=None):
+            return _Resp()
+
+    monkeypatch.setattr(
+        v2_port_bodies, "get_async_httpx_client", lambda **kw: _Client()
+    )
+    assert await resolve_v2_auth_value(_m2m_server()) == {
+        "Authorization": "Bearer m2m-tok"
+    }