Load conformance test credentials from MCP_CONFORMANCE_CONTEXT env var

Instead of hardcoding test credentials (private key, client_id, client_secret),
the conformance auth client now loads them from MCP_CONFORMANCE_CONTEXT
environment variable as a JSON object:

{
    "client_id": "...",
    "client_secret": "...",       // For client_secret_basic flow
    "private_key_pem": "...",     // For private_key_jwt flow
    "signing_algorithm": "ES256"  // Optional, defaults to ES256
}

This allows the conformance test framework to provide credentials dynamically.

Author: pcarleton

examples/clients/conformance-auth-client/mcp_conformance_auth_client/__init__.py

@@ -9,14 +9,25 @@
 Usage:
     python -m mcp_conformance_auth_client <scenario> <server-url>
 
+Environment Variables:
+    MCP_CONFORMANCE_CONTEXT - JSON object containing test credentials:
+        {
+            "client_id": "...",
+            "client_secret": "...",       # For client_secret_basic flow
+            "private_key_pem": "...",     # For private_key_jwt flow
+            "signing_algorithm": "ES256"  # Optional, defaults to ES256
+        }
+
 Scenarios:
     auth/*                      - Authorization code flow scenarios (default behavior)
     auth/client-credentials-jwt   - Client credentials with JWT authentication (SEP-1046)
     auth/client-credentials-basic - Client credentials with client_secret_basic
 """
 
 import asyncio
+import json
 import logging
+import os
 import sys
 from datetime import timedelta
 from urllib.parse import ParseResult, parse_qs, urlparse
@@ -29,20 +40,19 @@
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
 from pydantic import AnyUrl
 
-# Well-known test private key for conformance testing (EC P-256).
-# This corresponds to the public key in the conformance test framework.
-# The key is intentionally public for conformance testing purposes.
-CONFORMANCE_TEST_PRIVATE_KEY_PEM = """-----BEGIN EC PRIVATE KEY-----
-MHcCAQEEIAIbFaPsD3ShCQpvGzgyEIvLQo8pLOsEMwQWcRq+CwWGoAoGCCqGSM49
-AwEHoUQDQgAEiYsmK6YG68IEZ4k0vVSg1Cy8PRwsmz/XweRkTsYof+yVMH+FYm5P
-0IhAPaKV/4DjucO6UcQ+d/8q7i//mT7WQA==
------END EC PRIVATE KEY-----"""
-
-# The client_id expected by the conformance test
-CONFORMANCE_TEST_CLIENT_ID = "conformance-test-client"
 
-# Static credentials for client_secret_basic flow
-CONFORMANCE_CLIENT_SECRET = "conformance-test-secret"
+def get_conformance_context() -> dict:
+    """Load conformance test context from MCP_CONFORMANCE_CONTEXT environment variable."""
+    context_json = os.environ.get("MCP_CONFORMANCE_CONTEXT")
+    if not context_json:
+        raise RuntimeError(
+            "MCP_CONFORMANCE_CONTEXT environment variable not set. "
+            "Expected JSON with client_id, client_secret, and/or private_key_pem."
+        )
+    try:
+        return json.loads(context_json)
+    except json.JSONDecodeError as e:
+        raise RuntimeError(f"Failed to parse MCP_CONFORMANCE_CONTEXT as JSON: {e}") from e
 
 # Set up logging to stderr (stdout is for conformance test output)
 logging.basicConfig(
@@ -170,35 +180,50 @@ async def run_client_credentials_jwt_client(server_url: str) -> None:
 
     This function:
     1. Connects to the MCP server with OAuth client_credentials grant
-    2. Uses private_key_jwt authentication with the well-known test key
+    2. Uses private_key_jwt authentication with credentials from MCP_CONFORMANCE_CONTEXT
     3. Initializes the session
     4. Lists available tools
     5. Calls a test tool
     """
     logger.debug(f"Starting conformance auth client (client_credentials_jwt) for {server_url}")
 
+    # Load credentials from environment
+    context = get_conformance_context()
+    client_id = context.get("client_id")
+    private_key_pem = context.get("private_key_pem")
+    signing_algorithm = context.get("signing_algorithm", "ES256")
+
+    if not client_id:
+        raise RuntimeError("MCP_CONFORMANCE_CONTEXT missing 'client_id'")
+    if not private_key_pem:
+        raise RuntimeError("MCP_CONFORMANCE_CONTEXT missing 'private_key_pem'")
+
     # Create JWT parameters for private_key_jwt authentication
     jwt_params = JWTParameters(
-        issuer=CONFORMANCE_TEST_CLIENT_ID,
-        subject=CONFORMANCE_TEST_CLIENT_ID,
-        jwt_signing_algorithm="ES256",
-        jwt_signing_key=CONFORMANCE_TEST_PRIVATE_KEY_PEM,
+        issuer=client_id,
+        subject=client_id,
+        jwt_signing_algorithm=signing_algorithm,
+        jwt_signing_key=private_key_pem,
     )
 
     # Create OAuth authentication handler for client_credentials flow
     # Note: redirect_uris is required by the model but not used in client_credentials flow
-    oauth_auth = RFC7523OAuthClientProvider(
-        server_url=server_url,
-        client_metadata=OAuthClientMetadata(
-            client_name=CONFORMANCE_TEST_CLIENT_ID,
-            redirect_uris=[AnyUrl("http://localhost:0/unused")],  # Required but unused
-            grant_types=["client_credentials"],
-            response_types=[],
-            token_endpoint_auth_method="private_key_jwt",
-        ),
-        storage=InMemoryTokenStorage(),
-        jwt_parameters=jwt_params,
-    )
+    import warnings
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", DeprecationWarning)
+        oauth_auth = RFC7523OAuthClientProvider(
+            server_url=server_url,
+            client_metadata=OAuthClientMetadata(
+                client_name=client_id,
+                redirect_uris=[AnyUrl("http://localhost:0/unused")],  # Required but unused
+                grant_types=["client_credentials"],
+                response_types=[],
+                token_endpoint_auth_method="private_key_jwt",
+            ),
+            storage=InMemoryTokenStorage(),
+            jwt_parameters=jwt_params,
+        )
 
     await _run_session(server_url, oauth_auth)
 
@@ -209,36 +234,50 @@ async def run_client_credentials_basic_client(server_url: str) -> None:
 
     This function:
     1. Connects to the MCP server with OAuth client_credentials grant
-    2. Uses client_secret_basic authentication with static credentials
+    2. Uses client_secret_basic authentication with credentials from MCP_CONFORMANCE_CONTEXT
     3. Initializes the session
     4. Lists available tools
     5. Calls a test tool
     """
     logger.debug(f"Starting conformance auth client (client_credentials_basic) for {server_url}")
 
-    # Create storage pre-populated with static client credentials
+    # Load credentials from environment
+    context = get_conformance_context()
+    client_id = context.get("client_id")
+    client_secret = context.get("client_secret")
+
+    if not client_id:
+        raise RuntimeError("MCP_CONFORMANCE_CONTEXT missing 'client_id'")
+    if not client_secret:
+        raise RuntimeError("MCP_CONFORMANCE_CONTEXT missing 'client_secret'")
+
+    # Create storage pre-populated with client credentials
     storage = InMemoryTokenStorage()
     await storage.set_client_info(
         OAuthClientInformationFull(
-            client_id=CONFORMANCE_TEST_CLIENT_ID,
-            client_secret=CONFORMANCE_CLIENT_SECRET,
+            client_id=client_id,
+            client_secret=client_secret,
             redirect_uris=[AnyUrl("http://localhost:0/unused")],
             token_endpoint_auth_method="client_secret_basic",
         )
     )
 
     # Create OAuth authentication handler for client_credentials flow with basic auth
-    oauth_auth = RFC7523OAuthClientProvider(
-        server_url=server_url,
-        client_metadata=OAuthClientMetadata(
-            client_name=CONFORMANCE_TEST_CLIENT_ID,
-            redirect_uris=[AnyUrl("http://localhost:0/unused")],  # Required but unused
-            grant_types=["client_credentials"],
-            response_types=[],
-            token_endpoint_auth_method="client_secret_basic",
-        ),
-        storage=storage,
-    )
+    import warnings
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", DeprecationWarning)
+        oauth_auth = RFC7523OAuthClientProvider(
+            server_url=server_url,
+            client_metadata=OAuthClientMetadata(
+                client_name=client_id,
+                redirect_uris=[AnyUrl("http://localhost:0/unused")],  # Required but unused
+                grant_types=["client_credentials"],
+                response_types=[],
+                token_endpoint_auth_method="client_secret_basic",
+            ),
+            storage=storage,
+        )
 
     await _run_session(server_url, oauth_auth)