Anthropic MCP Registry API - Implementation Guide

Anthropic MCP Registry API - Implementation Guide

Note: The Anthropic API version (v0.1) is defined as a constant ANTHROPIC_API_VERSION in registry/constants.py. All code references this constant rather than hardcoding the version string.

Overview

This implementation provides full compatibility with the Anthropic MCP Registry REST API v0.1 specification, enabling seamless integration with MCP ecosystem tools and downstream applications.

Key Features

• ✅ 3 REST API endpoints for server discovery
• ✅ JWT Bearer token authentication via Keycloak
• ✅ Cursor-based pagination for server lists
• ✅ Permission-based filtering using MCP scopes
• ✅ Complete Pydantic models matching Anthropic spec
• ✅ Automatic data transformation from internal format

Architecture

┌─────────────────────────────────────────────────────────────┐
│ Client (Authorization: Bearer <JWT>)                        │
└────────────────────┬────────────────────────────────────────┘
                     │ HTTP Request
                     ▼
┌─────────────────────────────────────────────────────────────┐
│ Nginx (:80/:443)                                            │
│  └─ /v0.1/* location                                          │
│     └─ auth_request /validate  ────────────────┐            │
└────────────────────┬───────────────────────────┼────────────┘
                     │                            │
                     │                            ▼
                     │              ┌─────────────────────────┐
                     │              │ Auth Server (:8888)     │
                     │              │  - Validates JWT        │
                     │              │  - Checks Keycloak      │
                     │              │  - Returns headers      │
                     │              └─────────────┬───────────┘
                     │                            │
                     │ ◄──────────────────────────┘
                     │ X-User, X-Scopes, X-Username
                     ▼
┌─────────────────────────────────────────────────────────────┐
│ Registry FastAPI (:7860)                                    │
│  ├─ nginx_proxied_auth() - Reads headers                   │
│  ├─ registry_routes.py - API endpoints                           │
│  ├─ server_service - Data access                           │
│  └─ transform_service - Format conversion                  │
└────────────────────┬────────────────────────────────────────┘
                     │
                     ▼
            Anthropic Schema Response

File Structure

New Files

Modified Files

Constants Configuration

All hardcoded values are centralized in registry/constants.py:

class RegistryConstants(BaseModel):
    # Anthropic Registry API v0.1 constants
    ANTHROPIC_SERVER_NAMESPACE: str = "io.mcpgateway"
    ANTHROPIC_API_DEFAULT_LIMIT: int = 100
    ANTHROPIC_API_MAX_LIMIT: int = 1000

Usage: Import with from ..constants import REGISTRY_CONSTANTS

API Endpoints

1. List Servers

GET /v0.1/servers?cursor={cursor}&limit={limit}

Purpose: List all MCP servers the authenticated user can access.

Query Parameters:

• cursor (optional): Pagination cursor from previous response
• limit (optional): Results per page (1-1000, default 100)

Response: ServerList with pagination metadata

Example:

curl "http://localhost/v0.1/servers?limit=5" \
  -H "Authorization: Bearer $TOKEN"

2. List Server Versions

GET /v0.1/servers/{serverName:path}/versions

Purpose: List all available versions for a specific server.

URL Parameters:

• serverName: URL-encoded name (e.g., io.mcpgateway%2Ffininfo)

Response: ServerList (currently single version per server)

Important: Note :path route converter to handle / in server names.

Example:

curl "http://localhost/v0.1/servers/io.mcpgateway%2Ffininfo/versions" \
  -H "Authorization: Bearer $TOKEN"

3. Get Server Version Details

GET /v0.1/servers/{serverName:path}/versions/{version}

Purpose: Get detailed information for a specific server version.

URL Parameters:

• serverName: URL-encoded name (e.g., io.mcpgateway%2Ffininfo)
• version: Version string (use latest for current version)

Response: ServerResponse with full server details

Example:

curl "http://localhost/v0.1/servers/io.mcpgateway%2Ffininfo/versions/latest" \
  -H "Authorization: Bearer $TOKEN"

Authentication Flow

1. JWT Bearer Token Validation

Client → Nginx:

GET /v0.1/servers
Authorization: Bearer eyJhbGci...

Nginx → Auth Server (/validate endpoint):

GET /validate
X-Authorization: Bearer eyJhbGci...
X-Original-URL: http://localhost/v0.1/servers

Auth Server Processing:

1. Validates JWT signature using Keycloak JWKS
2. Checks expiration, issuer (3-tier validation), audience
   • Tries external URL: https://mcpgateway.ddns.net/realms/mcp-gateway
   • Tries internal URL: http://keycloak:8080/realms/mcp-gateway
   • Tries localhost URL: http://localhost:8080/realms/mcp-gateway
3. Extracts user info: preferred_username, groups, scope
4. Maps Keycloak groups to MCP scopes

Auth Server → Nginx (response headers):

X-User: service-account-mcp-gateway-m2m
X-Username: service-account-mcp-gateway-m2m
X-Scopes: mcp-servers-unrestricted/read mcp-servers-unrestricted/execute
X-Auth-Method: keycloak

Nginx → FastAPI:

GET /v0.1/servers
X-User: service-account-mcp-gateway-m2m
X-Scopes: mcp-servers-unrestricted/read mcp-servers-unrestricted/execute
Authorization: Bearer eyJhbGci...

2. nginx Configuration

Critical Setup in /v0.1/ location block:

location /v0.1/ {
    # Authenticate via auth-server
    auth_request /validate;

    # Capture auth server response headers
    auth_request_set $auth_user $upstream_http_x_user;
    auth_request_set $auth_username $upstream_http_x_username;
    auth_request_set $auth_scopes $upstream_http_x_scopes;
    auth_request_set $auth_method $upstream_http_x_auth_method;

    # Forward to FastAPI with auth context
    proxy_pass http://127.0.0.1:7860/v0.1/;
    proxy_set_header X-User $auth_user;
    proxy_set_header X-Username $auth_username;
    proxy_set_header X-Scopes $auth_scopes;
    proxy_set_header X-Auth-Method $auth_method;
    proxy_set_header Authorization $http_authorization;
}

Key Fix: /validate endpoint must forward Authorization as X-Authorization:

location = /validate {
    proxy_pass http://auth-server:8888/validate;
    # CRITICAL: Read from $http_authorization (client's Authorization header)
    proxy_set_header X-Authorization $http_authorization;
}

3. FastAPI Authentication Dependency

Function: nginx_proxied_auth() in registry/auth/dependencies.py

Supports Two Modes:

1. JWT Flow (primary): Reads nginx headers from auth validation
2. Cookie Flow (fallback): Reads session cookies for backward compatibility

def nginx_proxied_auth(
    request: Request,
    session: Cookie = None,
    x_user: Header = None,
    x_username: Header = None,
    x_scopes: Header = None,
    x_auth_method: Header = None,
) -> Dict[str, Any]:
    # Try nginx headers first (JWT Bearer token)
    if x_user or x_username:
        username = x_username or x_user
        scopes = x_scopes.split() if x_scopes else []

        # Map scopes to groups
        if 'mcp-servers-unrestricted/read' in scopes:
            groups = ['mcp-registry-admin']
        else:
            groups = ['mcp-registry-user']

        # Get accessible servers from scopes
        accessible_servers = get_user_accessible_servers(scopes)

        return {
            'username': username,
            'groups': groups,
            'scopes': scopes,
            'accessible_servers': accessible_servers,
            'is_admin': 'mcp-registry-admin' in groups,
            # ... more fields
        }

    # Fallback to session cookie
    return enhanced_auth(session)

Permission Checks

Scope-Based Access Control

IMPORTANT: v0.1 API uses accessible_servers (MCP scopes), NOT accessible_services (UI scopes).

# CORRECT - Check against accessible_servers
accessible_servers = user_context.get("accessible_servers", [])
if server_name not in accessible_servers:
    raise HTTPException(404, "Server not found")

Why:

• accessible_services = UI-level services ("auth_server", "mcpgw")
• accessible_servers = MCP server names ("fininfo", "currenttime")
• M2M tokens have MCP scopes but no UI scopes

User Context Structure

{
    "username": "service-account-mcp-gateway-m2m",
    "groups": ["mcp-registry-admin"],
    "scopes": [
        "mcp-servers-unrestricted/read",
        "mcp-servers-unrestricted/execute",
        "mcp-servers-restricted/read",
        "mcp-servers-restricted/execute"
    ],
    "auth_method": "keycloak",
    "provider": "keycloak",
    "accessible_servers": [
        "currenttime", "fininfo",
        "mcpgw", "realserverfaketools", "sre-gateway"
    ],
    "accessible_services": [],  # Empty for M2M tokens
    "is_admin": True,
    "can_modify_servers": False
}

Data Transformation

Namespace Convention

Internal Format: /fininfo, /currenttime/ Anthropic Format: io.mcpgateway/fininfo, io.mcpgateway/currenttime

Implementation (transform_service.py):

def _create_server_name(server_info: Dict[str, Any]) -> str:
    path = server_info.get("path", "")
    clean_path = path.strip("/")
    namespace = REGISTRY_CONSTANTS.ANTHROPIC_SERVER_NAMESPACE
    return f"{namespace}/{clean_path}"

Server Detail Transformation

def transform_to_server_detail(server_info: Dict[str, Any]) -> ServerDetail:
    # Create Anthropic-format name
    name = _create_server_name(server_info)

    # Build package with transport config
    transport = _create_transport_config(server_info)
    package = Package(
        registryType="mcpb",
        identifier=name,
        version="1.0.0",
        transport=transport,
        runtimeHint="docker"
    )

    # Add internal metadata
    namespace = REGISTRY_CONSTANTS.ANTHROPIC_SERVER_NAMESPACE
    meta = {
        f"{namespace}/internal": {
            "path": server_info.get("path"),
            "is_enabled": server_info.get("is_enabled"),
            "health_status": server_info.get("health_status"),
            "num_tools": server_info.get("num_tools"),
            "tags": server_info.get("tags", []),
            "license": server_info.get("license", "N/A")
        }
    }

    return ServerDetail(name=name, packages=[package], meta=meta, ...)

Response Structure

{
  "server": {
    "name": "io.mcpgateway/fininfo",
    "description": "Financial Information",
    "version": "1.0.0",
    "title": "Financial Info",
    "packages": [
      {
        "registryType": "mcpb",
        "identifier": "io.mcpgateway/fininfo",
        "version": "1.0.0",
        "transport": {
          "type": "streamable-http",
          "url": "http://fininfo-server:8001/mcp/"
        },
        "runtimeHint": "docker"
      }
    ],
    "_meta": {
      "io.mcpgateway/internal": {
        "path": "/fininfo",
        "is_enabled": true,
        "health_status": "healthy",
        "num_tools": 5,
        "tags": ["Finance", "Stocks", "Market"],
        "license": "MIT"
      }
    }
  },
  "_meta": {
    "io.mcpgateway/registry": {
      "last_checked": "2025-10-12T19:25:09.378358+00:00",
      "health_status": "healthy"
    }
  }
}

Pagination

Cursor-Based Implementation

Algorithm (transform_service.py):

def transform_to_server_list(
    servers_data: List[Dict[str, Any]],
    cursor: Optional[str] = None,
    limit: Optional[int] = None
) -> ServerList:
    # Apply defaults
    limit = limit or REGISTRY_CONSTANTS.ANTHROPIC_API_DEFAULT_LIMIT
    limit = min(limit, REGISTRY_CONSTANTS.ANTHROPIC_API_MAX_LIMIT)

    # Sort alphabetically for consistency
    sorted_servers = sorted(servers_data, key=lambda s: _create_server_name(s))

    # Find cursor position
    start_index = 0
    if cursor:
        for idx, server in enumerate(sorted_servers):
            if _create_server_name(server) == cursor:
                start_index = idx + 1
                break

    # Slice page
    end_index = start_index + limit
    page_servers = sorted_servers[start_index:end_index]

    # Determine next cursor
    has_more = end_index < len(sorted_servers)
    next_cursor = _create_server_name(sorted_servers[end_index - 1]) if has_more else None

    # Transform and return
    return ServerList(
        servers=[transform_to_server_response(s) for s in page_servers],
        metadata=PaginationMetadata(nextCursor=next_cursor, count=len(page_servers))
    )

Example Flow:

Page 1: GET /v0.1/servers?limit=3
← Returns: servers A, B, C with nextCursor="C"

Page 2: GET /v0.1/servers?cursor=C&limit=3
← Returns: servers D, E, F with nextCursor="F"

Page 3: GET /v0.1/servers?cursor=F&limit=3
← Returns: servers G, H with nextCursor=null (end)

Critical Implementation Details

1. Route Path Parameters

Problem: Server names contain / which breaks FastAPI routing.

Solution: Use :path converter in route definition.

# WRONG - Returns 404 for io.mcpgateway/fininfo
@router.get("/servers/{serverName}/versions")

# CORRECT - Captures full path including /
@router.get("/servers/{serverName:path}/versions")

Why: FastAPI URL-decodes before routing. io.mcpgateway%2Ffininfo becomes io.mcpgateway/fininfo, which looks like extra path segments without :path.

2. Trailing Slash Handling

Problem: Some servers have trailing slashes (/currenttime/), some don't (/fininfo).

Solution: Try both forms when looking up servers.

# Construct path from server name
lookup_path = "/" + decoded_name.replace(expected_prefix, "")

# Try with and without trailing slash
server_info = server_service.get_server_info(lookup_path)
if not server_info:
    server_info = server_service.get_server_info(lookup_path + "/")

# Use actual path from server_info for health checks
path = server_info.get("path", lookup_path)  # Has correct trailing slash
health_data = health_service._get_service_health_data(path)

Why: Health data is indexed by exact path. Wrong path returns "unknown" status.

3. Namespace Constant Usage

All occurrences of hardcoded "io.mcpgateway" replaced with constant:

from ..constants import REGISTRY_CONSTANTS

namespace = REGISTRY_CONSTANTS.ANTHROPIC_SERVER_NAMESPACE
expected_prefix = f"{namespace}/"  # "io.mcpgateway/"

Files using constant:

• registry/api/registry_routes.py - Validates server name format
• registry/services/transform_service.py - Creates names and metadata keys

Testing

Generate Token

# Generate fresh credentials (tokens expire after 5 minutes)
./generate_creds.sh

# Load token
export TOKEN=$(jq -r '.access_token' .oauth-tokens/ingress.json)

# Verify token loaded
echo "Token: ${TOKEN:0:50}..."

Test Endpoints

# 1. List servers with pagination
curl "http://localhost/v0.1/servers?limit=5" \
  -H "Authorization: Bearer $TOKEN" | jq

# 2. List versions for a server (note %2F = /)
curl "http://localhost/v0.1/servers/io.mcpgateway%2Ffininfo/versions" \
  -H "Authorization: Bearer $TOKEN" | jq

# 3. Get specific version details
curl "http://localhost/v0.1/servers/io.mcpgateway%2Ffininfo/versions/latest" \
  -H "Authorization: Bearer $TOKEN" | jq

# 4. Test pagination
curl "http://localhost/v0.1/servers?limit=2" \
  -H "Authorization: Bearer $TOKEN" | jq '.metadata'
# Get nextCursor and use it:
curl "http://localhost/v0.1/servers?cursor=io.mcpgateway%2Fcurrenttime&limit=2" \
  -H "Authorization: Bearer $TOKEN" | jq

Comprehensive Test Suite

See docs/design/anthropic-api-test-commands.md for 20 test scenarios.

Common Issues & Solutions

Issue: 404 on versions endpoint

Symptom: GET /v0.1/servers/io.mcpgateway%2Ffininfo/versions returns 404

Cause: Missing :path in route parameter

Solution: Ensure route uses {serverName:path} not {serverName}

Issue: Health data shows "unknown"

Symptom: health_status: "unknown", last_checked: null

Cause: Trailing slash mismatch in path lookup

Solution: Use server_info.get("path") for health checks, not constructed path

Issue: Empty server list

Symptom: {"servers": [], "metadata": {"count": 0}}

Cause: Checking accessible_services instead of accessible_servers

Solution: Use user_context["accessible_servers"] for permission checks

Issue: 401 Unauthorized

Symptom: {"detail": "Token has expired"}

Cause: JWT token expired (5 minute lifetime)

Solution: Run ./generate_creds.sh to get fresh token

Issue: Token not forwarded

Symptom: Auth server logs show Authorization=False

Cause: nginx using $http_x_authorization instead of $http_authorization

Solution: Update /validate location to use $http_authorization

Schema Compliance

OpenAPI Spec: https://github.com/modelcontextprotocol/registry/blob/main/docs/reference/api/openapi.yaml

Pydantic Models (registry/schemas/anthropic_schema.py):

• ✅ ServerList - Paginated server list
• ✅ ServerResponse - Single server with metadata
• ✅ ServerDetail - Complete server information
• ✅ Package - Distribution package details
• ✅ Transport - Union of transport types
• ✅ Repository - Source code repository info
• ✅ PaginationMetadata - Cursor and count
• ✅ ErrorResponse - Error details

Field Aliases: Pydantic handles _meta fields with Field(alias="_meta")

Next Steps

1. ✅ JWT Authentication - Fully implemented
2. ✅ Permission Filtering - Uses MCP scopes
3. ✅ Health Data - Includes status and last checked
4. ✅ Pagination - Cursor-based with configurable limits
5. 🔄 Read-Only API Access - Optional: Create dedicated M2M client with minimal scopes (see .scratchpad/registry-api-readonly-access.md)
6. 🔄 Rate Limiting - Future: Add per-client rate limits
7. 🔄 Caching - Future: Cache server list responses

References

• Issue: #175 - Support Anthropic MCP Registry REST API v0
• OpenAPI Spec: https://github.com/modelcontextprotocol/registry/blob/main/docs/reference/api/openapi.yaml
• API Guide: https://github.com/modelcontextprotocol/registry/blob/main/docs/guides/consuming/use-rest-api.md
• Test Commands: anthropic-api-test-commands.md
• Progress Notes: .scratchpad/anthropic-api-v0-jwt-auth-progress.md