modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add conformance testing CI pipeline (#1915)

Browse source
This commit is contained in:
Max Isbey 2026-01-21 16:04:35 +00:00 • committed by GitHub
commit 213cf993ca
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 422 additions and 434 deletions
Download patch file Download diff file Expand all files Collapse all files

346
.github/actions/conformance/client.py vendored Normal file
View file

@ -0,0 +1,346 @@
"""MCP unified conformance test client.
This client is designed to work with the @modelcontextprotocol/conformance npm package.
It handles all conformance test scenarios via environment variables and CLI arguments.
Contract:
- MCP_CONFORMANCE_SCENARIO env var -> scenario name
- MCP_CONFORMANCE_CONTEXT env var -> optional JSON (for client-credentials scenarios)
- Server URL as last CLI argument (sys.argv[1])
- Must exit 0 within 30 seconds
Scenarios:
initialize - Connect, initialize, list tools, close
tools_call - Connect, call add_numbers(a=5, b=3), close
sse-retry - Connect, call test_reconnection, close
elicitation-sep1034-client-defaults - Elicitation with default accept callback
auth/client-credentials-jwt - Client credentials with private_key_jwt
auth/client-credentials-basic - Client credentials with client_secret_basic
auth/* - Authorization code flow (default for auth scenarios)
"""
import asyncio
import json
import logging
import os
import sys
from collections.abc import Callable, Coroutine
from typing import Any, cast
from urllib.parse import parse_qs, urlparse
import httpx
from pydantic import AnyUrl
from mcp import ClientSession, types
from mcp.client.auth import OAuthClientProvider, TokenStorage
from mcp.client.auth.extensions.client_credentials import (
ClientCredentialsOAuthProvider,
PrivateKeyJWTOAuthProvider,
SignedJWTParameters,
)
from mcp.client.streamable_http import streamable_http_client
from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
from mcp.shared.context import RequestContext
# Set up logging to stderr (stdout is for conformance test output)
logging.basicConfig(
level=logging.DEBUG,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
stream=sys.stderr,
)
logger = logging.getLogger(__name__)
# Type for async scenario handler functions
ScenarioHandler = Callable[[str], Coroutine[Any, None, None]]
# Registry of scenario handlers
HANDLERS: dict[str, ScenarioHandler] = {}
def register(name: str) -> Callable[[ScenarioHandler], ScenarioHandler]:
"""Register a scenario handler."""
def decorator(fn: ScenarioHandler) -> ScenarioHandler:
HANDLERS[name] = fn
return fn
return decorator
def get_conformance_context() -> dict[str, Any]:
"""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
class InMemoryTokenStorage(TokenStorage):
"""Simple in-memory token storage for conformance testing."""
def __init__(self) -> None:
self._tokens: OAuthToken | None = None
self._client_info: OAuthClientInformationFull | None = None
async def get_tokens(self) -> OAuthToken | None:
return self._tokens
async def set_tokens(self, tokens: OAuthToken) -> None:
self._tokens = tokens
async def get_client_info(self) -> OAuthClientInformationFull | None:
return self._client_info
async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
self._client_info = client_info
class ConformanceOAuthCallbackHandler:
"""OAuth callback handler that automatically fetches the authorization URL
and extracts the auth code, without requiring user interaction.
"""
def __init__(self) -> None:
self._auth_code: str | None = None
self._state: str | None = None
async def handle_redirect(self, authorization_url: str) -> None:
"""Fetch the authorization URL and extract the auth code from the redirect."""
logger.debug(f"Fetching authorization URL: {authorization_url}")
async with httpx.AsyncClient() as client:
response = await client.get(
authorization_url,
follow_redirects=False,
)
if response.status_code in (301, 302, 303, 307, 308):
location = cast(str, response.headers.get("location"))
if location:
redirect_url = urlparse(location)
query_params: dict[str, list[str]] = parse_qs(redirect_url.query)
if "code" in query_params:
self._auth_code = query_params["code"][0]
state_values = query_params.get("state")
self._state = state_values[0] if state_values else None
logger.debug(f"Got auth code from redirect: {self._auth_code[:10]}...")
return
else:
raise RuntimeError(f"No auth code in redirect URL: {location}")
else:
raise RuntimeError(f"No redirect location received from {authorization_url}")
else:
raise RuntimeError(f"Expected redirect response, got {response.status_code} from {authorization_url}")
async def handle_callback(self) -> tuple[str, str | None]:
"""Return the captured auth code and state."""
if self._auth_code is None:
raise RuntimeError("No authorization code available - was handle_redirect called?")
auth_code = self._auth_code
state = self._state
self._auth_code = None
self._state = None
return auth_code, state
# --- Scenario Handlers ---
@register("initialize")
async def run_initialize(server_url: str) -> None:
"""Connect, initialize, list tools, close."""
async with streamable_http_client(url=server_url) as (read_stream, write_stream, _):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
logger.debug("Initialized successfully")
await session.list_tools()
logger.debug("Listed tools successfully")
@register("tools_call")
async def run_tools_call(server_url: str) -> None:
"""Connect, initialize, list tools, call add_numbers(a=5, b=3), close."""
async with streamable_http_client(url=server_url) as (read_stream, write_stream, _):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
await session.list_tools()
result = await session.call_tool("add_numbers", {"a": 5, "b": 3})
logger.debug(f"add_numbers result: {result}")
@register("sse-retry")
async def run_sse_retry(server_url: str) -> None:
"""Connect, initialize, list tools, call test_reconnection, close."""
async with streamable_http_client(url=server_url) as (read_stream, write_stream, _):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
await session.list_tools()
result = await session.call_tool("test_reconnection", {})
logger.debug(f"test_reconnection result: {result}")
async def default_elicitation_callback(
context: RequestContext[ClientSession, Any], # noqa: ARG001
params: types.ElicitRequestParams,
) -> types.ElicitResult | types.ErrorData:
"""Accept elicitation and apply defaults from the schema (SEP-1034)."""
content: dict[str, str | int | float | bool | list[str] | None] = {}
# For form mode, extract defaults from the requested_schema
if isinstance(params, types.ElicitRequestFormParams):
schema = params.requested_schema
logger.debug(f"Elicitation schema: {schema}")
properties = schema.get("properties", {})
for prop_name, prop_schema in properties.items():
if "default" in prop_schema:
content[prop_name] = prop_schema["default"]
logger.debug(f"Applied defaults: {content}")
return types.ElicitResult(action="accept", content=content)
@register("elicitation-sep1034-client-defaults")
async def run_elicitation_defaults(server_url: str) -> None:
"""Connect with elicitation callback that applies schema defaults."""
async with streamable_http_client(url=server_url) as (read_stream, write_stream, _):
async with ClientSession(
read_stream, write_stream, elicitation_callback=default_elicitation_callback
) as session:
await session.initialize()
await session.list_tools()
result = await session.call_tool("test_client_elicitation_defaults", {})
logger.debug(f"test_client_elicitation_defaults result: {result}")
@register("auth/client-credentials-jwt")
async def run_client_credentials_jwt(server_url: str) -> None:
"""Client credentials flow with private_key_jwt authentication."""
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'")
jwt_params = SignedJWTParameters(
issuer=client_id,
subject=client_id,
signing_algorithm=signing_algorithm,
signing_key=private_key_pem,
)
oauth_auth = PrivateKeyJWTOAuthProvider(
server_url=server_url,
storage=InMemoryTokenStorage(),
client_id=client_id,
assertion_provider=jwt_params.create_assertion_provider(),
)
await _run_auth_session(server_url, oauth_auth)
@register("auth/client-credentials-basic")
async def run_client_credentials_basic(server_url: str) -> None:
"""Client credentials flow with client_secret_basic authentication."""
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'")
oauth_auth = ClientCredentialsOAuthProvider(
server_url=server_url,
storage=InMemoryTokenStorage(),
client_id=client_id,
client_secret=client_secret,
token_endpoint_auth_method="client_secret_basic",
)
await _run_auth_session(server_url, oauth_auth)
async def run_auth_code_client(server_url: str) -> None:
"""Authorization code flow (default for auth/* scenarios)."""
callback_handler = ConformanceOAuthCallbackHandler()
oauth_auth = OAuthClientProvider(
server_url=server_url,
client_metadata=OAuthClientMetadata(
client_name="conformance-client",
redirect_uris=[AnyUrl("http://localhost:3000/callback")],
grant_types=["authorization_code", "refresh_token"],
response_types=["code"],
),
storage=InMemoryTokenStorage(),
redirect_handler=callback_handler.handle_redirect,
callback_handler=callback_handler.handle_callback,
client_metadata_url="https://conformance-test.local/client-metadata.json",
)
await _run_auth_session(server_url, oauth_auth)
async def _run_auth_session(server_url: str, oauth_auth: OAuthClientProvider) -> None:
"""Common session logic for all OAuth flows."""
client = httpx.AsyncClient(auth=oauth_auth, timeout=30.0)
async with streamable_http_client(url=server_url, http_client=client) as (read_stream, write_stream, _):
async with ClientSession(
read_stream, write_stream, elicitation_callback=default_elicitation_callback
) as session:
await session.initialize()
logger.debug("Initialized successfully")
tools_result = await session.list_tools()
logger.debug(f"Listed tools: {[t.name for t in tools_result.tools]}")
# Call the first available tool (different tests have different tools)
if tools_result.tools:
tool_name = tools_result.tools[0].name
try:
result = await session.call_tool(tool_name, {})
logger.debug(f"Called {tool_name}, result: {result}")
except Exception as e:
logger.debug(f"Tool call result/error: {e}")
logger.debug("Connection closed successfully")
def main() -> None:
"""Main entry point for the conformance client."""
if len(sys.argv) < 2:
print(f"Usage: {sys.argv[0]} <server-url>", file=sys.stderr)
sys.exit(1)
server_url = sys.argv[1]
scenario = os.environ.get("MCP_CONFORMANCE_SCENARIO")
if scenario:
logger.debug(f"Running explicit scenario '{scenario}' against {server_url}")
handler = HANDLERS.get(scenario)
if handler:
asyncio.run(handler(server_url))
elif scenario.startswith("auth/"):
asyncio.run(run_auth_code_client(server_url))
else:
print(f"Unknown scenario: {scenario}", file=sys.stderr)
sys.exit(1)
else:
logger.debug(f"Running default auth flow against {server_url}")
asyncio.run(run_auth_code_client(server_url))
if __name__ == "__main__":
main()

30
.github/actions/conformance/run-server.sh vendored Executable file
View file

@ -0,0 +1,30 @@
#!/bin/bash
set -e
PORT="${PORT:-3001}"
SERVER_URL="http://localhost:${PORT}/mcp"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
cd "$SCRIPT_DIR/../../.."
# Start everything-server
uv run --frozen mcp-everything-server --port "$PORT" &
SERVER_PID=$!
trap "kill $SERVER_PID 2>/dev/null || true; wait $SERVER_PID 2>/dev/null || true" EXIT
# Wait for server to be ready
MAX_RETRIES=30
RETRY_COUNT=0
while ! curl -s "$SERVER_URL" > /dev/null 2>&1; do
RETRY_COUNT=$((RETRY_COUNT + 1))
if [ $RETRY_COUNT -ge $MAX_RETRIES ]; then
echo "Server failed to start after ${MAX_RETRIES} retries" >&2
exit 1
fi
sleep 0.5
done
echo "Server ready at $SERVER_URL"
# Run conformance tests
npx @modelcontextprotocol/conformance@0.1.10 server --url "$SERVER_URL" "$@"

45
.github/workflows/conformance.yml vendored Normal file
View file

@ -0,0 +1,45 @@
name: Conformance Tests
on:
push:
branches: [main]
pull_request:
workflow_dispatch:
concurrency:
group: conformance-${{ github.ref }}
cancel-in-progress: true
permissions:
contents: read
jobs:
server-conformance:
runs-on: ubuntu-latest
continue-on-error: true
steps:
- uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1
- uses: astral-sh/setup-uv@61cb8a9741eeb8a550a1b8544337180c0fc8476b # v7.2.0
with:
enable-cache: true
version: 0.9.5
- uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0
with:
node-version: 24
- run: uv sync --frozen --all-extras --package mcp-everything-server
- run: ./.github/actions/conformance/run-server.sh
client-conformance:
runs-on: ubuntu-latest
continue-on-error: true
steps:
- uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1
- uses: astral-sh/setup-uv@61cb8a9741eeb8a550a1b8544337180c0fc8476b # v7.2.0
with:
enable-cache: true
version: 0.9.5
- uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0
with:
node-version: 24
- run: uv sync --frozen --all-extras --package mcp
- run: npx @modelcontextprotocol/conformance@0.1.10 client --command 'uv run --frozen python .github/actions/conformance/client.py' --suite all

1
.gitignore vendored
View file

@ -172,3 +172,4 @@ cython_debug/
# claude code
.claude/
results/

49
examples/clients/conformance-auth-client/README.md
View file

@ -1,49 +0,0 @@
# MCP Conformance Auth Client
A Python OAuth client designed for use with the MCP conformance test framework.
## Overview
This client implements OAuth authentication for MCP and is designed to work automatically with the conformance test framework without requiring user interaction. It programmatically fetches authorization URLs and extracts auth codes from redirects.
## Installation
```bash
cd examples/clients/conformance-auth-client
uv sync
```
## Usage with Conformance Tests
Run the auth conformance tests against this Python client:
```bash
# From the conformance repository
npx @modelcontextprotocol/conformance client \
--command "uv run --directory /path/to/python-sdk/examples/clients/conformance-auth-client python -m mcp_conformance_auth_client" \
--scenario auth/basic-dcr
```
Available auth test scenarios:
- `auth/basic-dcr` - Tests OAuth Dynamic Client Registration flow
- `auth/basic-metadata-var1` - Tests OAuth with authorization metadata
## How It Works
Unlike interactive OAuth clients that open a browser for user authentication, this client:
1. Receives the authorization URL from the OAuth provider
2. Makes an HTTP request to that URL directly (without following redirects)
3. Extracts the authorization code from the redirect response
4. Uses the code to complete the OAuth token exchange
This allows the conformance test framework's mock OAuth server to automatically provide auth codes without human interaction.
## Direct Usage
You can also run the client directly:
```bash
uv run python -m mcp_conformance_auth_client http://localhost:3000/mcp
```

306
examples/clients/conformance-auth-client/mcp_conformance_auth_client/__init__.py
View file

@ -1,306 +0,0 @@
#!/usr/bin/env python3
"""MCP OAuth conformance test client.
This client is designed to work with the MCP conformance test framework.
It automatically handles OAuth flows without user interaction by programmatically
fetching the authorization URL and extracting the auth code from the redirect.
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 typing import Any, cast
from urllib.parse import parse_qs, urlparse
import httpx
from mcp import ClientSession
from mcp.client.auth import OAuthClientProvider, TokenStorage
from mcp.client.auth.extensions.client_credentials import (
ClientCredentialsOAuthProvider,
PrivateKeyJWTOAuthProvider,
SignedJWTParameters,
)
from mcp.client.streamable_http import streamable_http_client
from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
from pydantic import AnyUrl
def get_conformance_context() -> dict[str, Any]:
"""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(
level=logging.DEBUG,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
stream=sys.stderr,
)
logger = logging.getLogger(__name__)
class InMemoryTokenStorage(TokenStorage):
"""Simple in-memory token storage for conformance testing."""
def __init__(self):
self._tokens: OAuthToken | None = None
self._client_info: OAuthClientInformationFull | None = None
async def get_tokens(self) -> OAuthToken | None:
return self._tokens
async def set_tokens(self, tokens: OAuthToken) -> None:
self._tokens = tokens
async def get_client_info(self) -> OAuthClientInformationFull | None:
return self._client_info
async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
self._client_info = client_info
class ConformanceOAuthCallbackHandler:
"""OAuth callback handler that automatically fetches the authorization URL
and extracts the auth code, without requiring user interaction.
This mimics the behavior of the TypeScript ConformanceOAuthProvider.
"""
def __init__(self):
self._auth_code: str | None = None
self._state: str | None = None
async def handle_redirect(self, authorization_url: str) -> None:
"""Fetch the authorization URL and extract the auth code from the redirect.
The conformance test server returns a redirect with the auth code,
so we can capture it programmatically.
"""
logger.debug(f"Fetching authorization URL: {authorization_url}")
async with httpx.AsyncClient() as client:
response = await client.get(
authorization_url,
follow_redirects=False, # Don't follow redirects automatically
)
# Check for redirect response
if response.status_code in (301, 302, 303, 307, 308):
location = cast(str, response.headers.get("location"))
if location:
redirect_url = urlparse(location)
query_params: dict[str, list[str]] = parse_qs(redirect_url.query)
if "code" in query_params:
self._auth_code = query_params["code"][0]
state_values = query_params.get("state")
self._state = state_values[0] if state_values else None
logger.debug(f"Got auth code from redirect: {self._auth_code[:10]}...")
return
else:
raise RuntimeError(f"No auth code in redirect URL: {location}")
else:
raise RuntimeError(f"No redirect location received from {authorization_url}")
else:
raise RuntimeError(f"Expected redirect response, got {response.status_code} from {authorization_url}")
async def handle_callback(self) -> tuple[str, str | None]:
"""Return the captured auth code and state, then clear them for potential reuse."""
if self._auth_code is None:
raise RuntimeError("No authorization code available - was handle_redirect called?")
auth_code = self._auth_code
state = self._state
# Clear the stored values so the next auth flow gets fresh ones
self._auth_code = None
self._state = None
return auth_code, state
async def run_authorization_code_client(server_url: str) -> None:
"""Run the conformance test client with authorization code flow.
This function:
1. Connects to the MCP server with OAuth authorization code flow
2. Initializes the session
3. Lists available tools
4. Calls a test tool
"""
logger.debug(f"Starting conformance auth client (authorization_code) for {server_url}")
# Create callback handler that will automatically fetch auth codes
callback_handler = ConformanceOAuthCallbackHandler()
# Create OAuth authentication handler
oauth_auth = OAuthClientProvider(
server_url=server_url,
client_metadata=OAuthClientMetadata(
client_name="conformance-auth-client",
redirect_uris=[AnyUrl("http://localhost:3000/callback")],
grant_types=["authorization_code", "refresh_token"],
response_types=["code"],
),
storage=InMemoryTokenStorage(),
redirect_handler=callback_handler.handle_redirect,
callback_handler=callback_handler.handle_callback,
)
await _run_session(server_url, oauth_auth)
async def run_client_credentials_jwt_client(server_url: str) -> None:
"""Run the conformance test client with client credentials flow using private_key_jwt (SEP-1046).
This function:
1. Connects to the MCP server with OAuth client_credentials grant
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 SDK-signed assertions
jwt_params = SignedJWTParameters(
issuer=client_id,
subject=client_id,
signing_algorithm=signing_algorithm,
signing_key=private_key_pem,
)
# Create OAuth provider for client_credentials with private_key_jwt
oauth_auth = PrivateKeyJWTOAuthProvider(
server_url=server_url,
storage=InMemoryTokenStorage(),
client_id=client_id,
assertion_provider=jwt_params.create_assertion_provider(),
)
await _run_session(server_url, oauth_auth)
async def run_client_credentials_basic_client(server_url: str) -> None:
"""Run the conformance test client with client credentials flow using client_secret_basic.
This function:
1. Connects to the MCP server with OAuth client_credentials grant
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}")
# 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 OAuth provider for client_credentials with client_secret_basic
oauth_auth = ClientCredentialsOAuthProvider(
server_url=server_url,
storage=InMemoryTokenStorage(),
client_id=client_id,
client_secret=client_secret,
token_endpoint_auth_method="client_secret_basic",
)
await _run_session(server_url, oauth_auth)
async def _run_session(server_url: str, oauth_auth: OAuthClientProvider) -> None:
"""Common session logic for all OAuth flows."""
# Connect using streamable HTTP transport with OAuth
client = httpx.AsyncClient(auth=oauth_auth, timeout=30.0)
async with streamable_http_client(url=server_url, http_client=client) as (read_stream, write_stream, _):
async with ClientSession(read_stream, write_stream) as session:
# Initialize the session
await session.initialize()
logger.debug("Successfully connected and initialized MCP session")
# List tools
tools_result = await session.list_tools()
logger.debug(f"Listed tools: {[t.name for t in tools_result.tools]}")
# Call test tool (expected by conformance tests)
try:
result = await session.call_tool("test-tool", {})
logger.debug(f"Called test-tool, result: {result}")
except Exception as e:
logger.debug(f"Tool call result/error: {e}")
logger.debug("Connection closed successfully")
def main() -> None:
"""Main entry point for the conformance auth client."""
if len(sys.argv) != 3:
print(f"Usage: {sys.argv[0]} <scenario> <server-url>", file=sys.stderr)
print("", file=sys.stderr)
print("Scenarios:", file=sys.stderr)
print(" auth/* - Authorization code flow (default)", file=sys.stderr)
print(" auth/client-credentials-jwt - Client credentials with JWT auth (SEP-1046)", file=sys.stderr)
print(" auth/client-credentials-basic - Client credentials with client_secret_basic", file=sys.stderr)
sys.exit(1)
scenario = sys.argv[1]
server_url = sys.argv[2]
try:
if scenario == "auth/client-credentials-jwt":
asyncio.run(run_client_credentials_jwt_client(server_url))
elif scenario == "auth/client-credentials-basic":
asyncio.run(run_client_credentials_basic_client(server_url))
else:
# Default to authorization code flow for all other auth/* scenarios
asyncio.run(run_authorization_code_client(server_url))
except Exception:
logger.exception("Client failed")
sys.exit(1)
if __name__ == "__main__":
main()

6
examples/clients/conformance-auth-client/mcp_conformance_auth_client/__main__.py
View file

@ -1,6 +0,0 @@
"""Allow running the module with python -m."""
from . import main
if __name__ == "__main__":
main()

43
examples/clients/conformance-auth-client/pyproject.toml
View file

@ -1,43 +0,0 @@
[project]
name = "mcp-conformance-auth-client"
version = "0.1.0"
description = "OAuth conformance test client for MCP"
readme = "README.md"
requires-python = ">=3.10"
authors = [{ name = "Anthropic" }]
keywords = ["mcp", "oauth", "client", "auth", "conformance", "testing"]
license = { text = "MIT" }
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.10",
]
dependencies = ["mcp", "httpx>=0.28.1"]
[project.scripts]
mcp-conformance-auth-client = "mcp_conformance_auth_client:main"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["mcp_conformance_auth_client"]
[tool.pyright]
include = ["mcp_conformance_auth_client"]
venvPath = "."
venv = ".venv"
[tool.ruff.lint]
select = ["E", "F", "I"]
ignore = []
[tool.ruff]
line-length = 120
target-version = "py310"
[dependency-groups]
dev = ["pyright>=1.1.379", "pytest>=8.3.3", "ruff>=0.6.9"]

30
uv.lock generated
View file

@ -9,7 +9,6 @@ resolution-markers = [
[manifest]
members = [
"mcp",
"mcp-conformance-auth-client",
"mcp-everything-server",
"mcp-simple-auth",
"mcp-simple-auth-client",
@ -822,35 +821,6 @@ docs = [
{ name = "mkdocstrings-python", specifier = ">=2.0.1" },
]
[[package]]
name = "mcp-conformance-auth-client"
version = "0.1.0"
source = { editable = "examples/clients/conformance-auth-client" }
dependencies = [
{ name = "httpx" },
{ name = "mcp" },
]
[package.dev-dependencies]
dev = [
{ name = "pyright" },
{ name = "pytest" },
{ name = "ruff" },
]
[package.metadata]
requires-dist = [
{ name = "httpx", specifier = ">=0.28.1" },
{ name = "mcp", editable = "." },
]
[package.metadata.requires-dev]
dev = [
{ name = "pyright", specifier = ">=1.1.379" },
{ name = "pytest", specifier = ">=8.3.3" },
{ name = "ruff", specifier = ">=0.6.9" },
]
[[package]]
name = "mcp-everything-server"
version = "0.1.0"