消费 Registry 数据

消费 Registry 数据

通过 REST API 构建消费 MCP Registry 数据的应用程序的集成模式和最佳实践。

基本信息

基础 URL: https://registry.modelcontextprotocol.io

身份验证: 只读访问无需身份验证

核心端点:

  • GET /v0/servers - 分页列出所有服务器
  • GET /v0/servers/{id} - 按 UUID 获取服务器详情

查看 交互式 API 文档 了解完整的请求/响应模式。

免责声明: 官方 Registry 不提供正常运行时间或数据持久性保证。您应该设计应用程序通过缓存来处理服务中断。

快速开始

基本服务器列表 

# 获取前 10 个服务器
curl "https://registry.modelcontextprotocol.io/v0/servers?limit=10"
{
  "servers": [
    {
      "name": "io.modelcontextprotocol/filesystem",
      "description": "文件系统操作服务器", 
      "status": "active",
      "version": "1.0.2"
    }
  ],
  "metadata": {
    "count": 10,
    "next_cursor": "eyJ..."
  }
}

搜索服务器 

# 搜索特定功能
curl "https://registry.modelcontextprotocol.io/v0/servers?search=filesystem"

# 按状态过滤
curl "https://registry.modelcontextprotocol.io/v0/servers?status=active"

# 组合查询
curl "https://registry.modelcontextprotocol.io/v0/servers?search=weather&limit=5"

获取服务器详情 

# 获取特定服务器的完整信息
curl "https://registry.modelcontextprotocol.io/v0/servers/{server-uuid}"

构建子注册中心

创建增强的注册中心 - ETL 官方 Registry 数据并添加您自己的元数据，如评级、安全扫描或兼容性信息。

ETL 流程

目前我们建议定期抓取 GET /v0/servers 端点。未来我们可能会提供 updated_at 过滤器（#291）以仅获取最近更改的服务器。

import requests
import time
from datetime import datetime

class RegistryETL:
    def __init__(self, base_url="https://registry.modelcontextprotocol.io"):
        self.base_url = base_url
        
    def fetch_all_servers(self):
        """获取所有服务器，处理分页"""
        servers = []
        cursor = None
        
        while True:
            params = {"limit": 100}
            if cursor:
                params["cursor"] = cursor
                
            response = requests.get(f"{self.base_url}/v0/servers", params=params)
            response.raise_for_status()
            
            data = response.json()
            servers.extend(data["servers"])
            
            # 检查是否有更多页面
            if not data.get("metadata", {}).get("next_cursor"):
                break
                
            cursor = data["metadata"]["next_cursor"]
            
        return servers
    
    def process_servers(self, servers):
        """处理服务器数据并添加增强信息"""
        enhanced_servers = []
        
        for server in servers:
            # 跳过非活跃服务器
            if server.get("status") != "active":
                continue
                
            # 添加自定义元数据
            enhanced_server = {
                **server,
                "_meta": {
                    **server.get("_meta", {}),
                    "com.yourregistry/enhanced": {
                        "last_processed": datetime.utcnow().isoformat(),
                        "popularity_score": self.calculate_popularity(server),
                        "security_scan": self.perform_security_scan(server)
                    }
                }
            }
            
            enhanced_servers.append(enhanced_server)
            
        return enhanced_servers
    
    def calculate_popularity(self, server):
        """计算服务器受欢迎程度（示例逻辑）"""
        # 实现您的评分逻辑
        base_score = 1.0
        
        # 基于描述质量调整
        if len(server.get("description", "")) > 100:
            base_score += 0.2
            
        # 基于包数量调整
        package_count = len(server.get("packages", []))
        if package_count > 1:
            base_score += 0.1 * package_count
            
        return min(base_score, 5.0)
    
    def perform_security_scan(self, server):
        """执行安全扫描（示例）"""
        return {
            "last_scanned": datetime.utcnow().isoformat(),
            "vulnerabilities_found": 0,
            "scan_status": "passed"
        }

# 使用示例
etl = RegistryETL()
servers = etl.fetch_all_servers()
enhanced_servers = etl.process_servers(servers)

状态管理

服务器通常是不可变的，除了 status 字段可以更新为 deleted（以及其他状态）。对于这些包，我们建议您也快速将状态字段更新为 deleted 或从您的注册中心中删除包。这是因为此状态通常表明它违反了我们宽松的 审核指南，表明它是非法的、恶意软件或垃圾邮件。

def sync_server_status(self, local_server, remote_server):
    """同步服务器状态变更"""
    if remote_server["status"] == "deleted":
        # 立即从本地注册中心删除或标记
        self.mark_server_deleted(local_server["id"])
        
    elif remote_server["status"] != local_server.get("status"):
        # 更新其他状态变更
        self.update_server_status(
            local_server["id"], 
            remote_server["status"]
        )

过滤和增强

官方 Registry 有 宽松的审核政策，因此您可能希望在 Registry 数据基础上实施自己的过滤。

def apply_quality_filters(self, servers):
    """应用质量过滤器"""
    filtered_servers = []
    
    for server in servers:
        # 基本质量检查
        if not self.meets_quality_standards(server):
            continue
            
        # 安全检查
        if not self.passes_security_scan(server):
            continue
            
        # 许可证检查
        if not self.has_acceptable_license(server):
            continue
            
        filtered_servers.append(server)
        
    return filtered_servers

def enhance_with_metadata(self, server):
    """为服务器添加增强元数据"""
    enhanced = {
        **server,
        "_meta": {
            **server.get("_meta", {}),
            "com.yourregistry/enhanced": {
                "user_rating": self.get_user_rating(server["name"]),
                "download_count": self.get_download_count(server["name"]),
                "last_updated": self.get_last_update_time(server["name"]),
                "tags": self.generate_tags(server),
                "compatibility": self.check_compatibility(server)
            }
        }
    }
    
    return enhanced

提供 API

我们建议您的子注册中心提供符合 Registry API 规范的 API，以便客户端可以轻松在注册中心之间切换。参见 Registry API 文档 了解详情。

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/v0/servers')
def list_servers():
    """实现与官方 Registry 兼容的服务器列表端点"""
    
    # 解析查询参数
    limit = min(int(request.args.get('limit', 20)), 100)
    cursor = request.args.get('cursor')
    search = request.args.get('search')
    status = request.args.get('status', 'active')
    
    # 应用过滤器
    servers = self.get_filtered_servers(
        limit=limit,
        cursor=cursor,
        search=search,
        status=status
    )
    
    # 计算下一页游标
    next_cursor = None
    if len(servers) == limit:
        next_cursor = self.generate_cursor(servers[-1])
    
    return jsonify({
        "servers": servers,
        "metadata": {
            "count": len(servers),
            "next_cursor": next_cursor
        }
    })

@app.route('/v0/servers/<server_id>')
def get_server(server_id):
    """获取特定服务器详情"""
    server = self.get_server_by_id(server_id)
    
    if not server:
        return jsonify({"error": "Server not found"}), 404
        
    return jsonify(server)

MCP 客户端集成

将 Registry 数据转换为客户端配置 - 获取服务器并将包信息转换为您的 MCP 客户端的配置格式。

基本集成 

import requests
from typing import List, Dict, Any

class MCPRegistryClient:
    def __init__(self, registry_url="https://registry.modelcontextprotocol.io"):
        self.registry_url = registry_url
        
    def discover_servers(self, search_query: str = None) -> List[Dict[str, Any]]:
        """发现可用的 MCP 服务器"""
        params = {"status": "active", "limit": 50}
        if search_query:
            params["search"] = search_query
            
        response = requests.get(f"{self.registry_url}/v0/servers", params=params)
        response.raise_for_status()
        
        return response.json()["servers"]
    
    def get_server_details(self, server_id: str) -> Dict[str, Any]:
        """获取服务器详细信息"""
        response = requests.get(f"{self.registry_url}/v0/servers/{server_id}")
        response.raise_for_status()
        
        return response.json()
    
    def convert_to_client_config(self, server: Dict[str, Any]) -> Dict[str, Any]:
        """将 Registry 服务器信息转换为客户端配置"""
        
        if "packages" in server:
            return self.convert_package_server(server)
        elif "remotes" in server:
            return self.convert_remote_server(server)
        else:
            raise ValueError(f"Server {server['name']} has no packages or remotes")
    
    def convert_package_server(self, server: Dict[str, Any]) -> Dict[str, Any]:
        """转换包类型服务器"""
        # 选择首选包（例如优先选择 npm）
        package = self.select_preferred_package(server["packages"])
        
        config = {
            "name": server["name"],
            "description": server["description"],
            "type": "package",
            "package": {
                "registry": package["registry_type"],
                "identifier": package["identifier"],
                "version": package["version"]
            }
        }
        
        # 添加运行时参数
        if "package_arguments" in package:
            config["package"]["arguments"] = package["package_arguments"]
            
        if "environment_variables" in package:
            config["package"]["environment"] = package["environment_variables"]
            
        return config
    
    def convert_remote_server(self, server: Dict[str, Any]) -> Dict[str, Any]:
        """转换远程类型服务器"""
        # 选择首选传输（例如优先选择 SSE）
        remote = self.select_preferred_remote(server["remotes"])
        
        config = {
            "name": server["name"],
            "description": server["description"],
            "type": "remote",
            "remote": {
                "transport": remote["type"],
                "url": remote["url"]
            }
        }
        
        # 添加头部
        if "headers" in remote:
            config["remote"]["headers"] = remote["headers"]
            
        return config
    
    def select_preferred_package(self, packages: List[Dict[str, Any]]) -> Dict[str, Any]:
        """选择首选包类型"""
        # 优先级顺序
        priority_order = ["npm", "pypi", "nuget", "oci", "mcpb"]
        
        for registry_type in priority_order:
            for package in packages:
                if package["registry_type"] == registry_type:
                    return package
                    
        # 如果没有找到首选类型，返回第一个
        return packages[0]
    
    def select_preferred_remote(self, remotes: List[Dict[str, Any]]) -> Dict[str, Any]:
        """选择首选远程传输"""
        # 优先选择 SSE
        for remote in remotes:
            if remote["type"] == "sse":
                return remote
                
        # 回退到第一个可用的
        return remotes[0]

# 使用示例
client = MCPRegistryClient()

# 发现服务器
servers = client.discover_servers("filesystem")

# 转换为客户端配置
configs = []
for server in servers:
    try:
        config = client.convert_to_client_config(server)
        configs.append(config)
    except ValueError as e:
        print(f"跳过服务器 {server['name']}: {e}")

print(f"找到 {len(configs)} 个可用服务器")

高级过滤

我们强烈建议使用子注册中心而不是直接从官方 Registry 获取数据。您可能希望使其可配置，以便客户端用户可以选择他们首选的注册中心，例如我们预期一些企业用户可能有自己的注册中心。

class AdvancedMCPClient:
    def __init__(self, preferred_registries=None):
        self.preferred_registries = preferred_registries or [
            "https://registry.modelcontextprotocol.io",
            "https://enterprise.registry.com/api/v0",
            "https://curated.mcpregistry.org/api/v0"
        ]
    
    def discover_servers_multi_registry(self, query: str) -> List[Dict[str, Any]]:
        """从多个注册中心发现服务器"""
        all_servers = []
        
        for registry_url in self.preferred_registries:
            try:
                client = MCPRegistryClient(registry_url)
                servers = client.discover_servers(query)
                
                # 添加来源信息
                for server in servers:
                    server["_source_registry"] = registry_url
                    
                all_servers.extend(servers)
                
            except Exception as e:
                print(f"从 {registry_url} 获取失败: {e}")
                continue
        
        # 去重（基于服务器名称）
        seen = set()
        unique_servers = []
        
        for server in all_servers:
            if server["name"] not in seen:
                seen.add(server["name"])
                unique_servers.append(server)
                
        return unique_servers
    
    def filter_by_criteria(self, servers: List[Dict[str, Any]], criteria: Dict[str, Any]) -> List[Dict[str, Any]]:
        """根据标准过滤服务器"""
        filtered = []
        
        for server in servers:
            # 状态过滤
            if criteria.get("status") and server.get("status") != criteria["status"]:
                continue
                
            # 包类型过滤
            if criteria.get("package_types"):
                server_types = {pkg["registry_type"] for pkg in server.get("packages", [])}
                if not server_types.intersection(set(criteria["package_types"])):
                    continue
            
            # 评分过滤
            if criteria.get("min_rating"):
                rating = server.get("_meta", {}).get("user_rating", 0)
                if rating < criteria["min_rating"]:
                    continue
                    
            filtered.append(server)
            
        return filtered

您的客户端应该优雅地处理满足最低规范的注册中心，即避免对 _meta 字段的硬依赖。

运行服务器

您可以使用 packages 或 remotes 字段来确定如何运行服务器。这些字段的更多详细信息在 server.json 文档 中。

def execute_server(self, config: Dict[str, Any]) -> subprocess.Popen:
    """执行 MCP 服务器"""
    
    if config["type"] == "package":
        return self.execute_package_server(config)
    elif config["type"] == "remote":
        return self.connect_remote_server(config)
    else:
        raise ValueError(f"不支持的服务器类型: {config['type']}")

def execute_package_server(self, config: Dict[str, Any]) -> subprocess.Popen:
    """执行包类型服务器"""
    package = config["package"]
    
    # 构建命令
    if package["registry"] == "npm":
        cmd = ["npx", package["identifier"]]
    elif package["registry"] == "pypi":
        cmd = ["uvx" if "runtime_hint" in package else "python", "-m", package["identifier"]]
    elif package["registry"] == "oci":
        cmd = ["docker", "run", package["identifier"]]
    else:
        raise ValueError(f"不支持的包注册中心: {package['registry']}")
    
    # 添加参数
    if "arguments" in package:
        for arg in package["arguments"]:
            if arg["type"] == "positional":
                cmd.append(arg["value"])
            elif arg["type"] == "named":
                cmd.extend([arg["name"], arg["value"]])
    
    # 设置环境变量
    env = os.environ.copy()
    if "environment" in package:
        for env_var in package["environment"]:
            if env_var.get("is_required") and env_var["name"] not in env:
                raise ValueError(f"缺少必需的环境变量: {env_var['name']}")
            env[env_var["name"]] = env.get(env_var["name"], env_var.get("default", ""))
    
    return subprocess.Popen(cmd, env=env, stdin=subprocess.PIPE, stdout=subprocess.PIPE)

最佳实践

1. 缓存策略 

import redis
import json
from datetime import timedelta

class CachedRegistryClient:
    def __init__(self, redis_client=None, cache_ttl=3600):
        self.redis = redis_client or redis.Redis()
        self.cache_ttl = cache_ttl
    
    def get_servers_cached(self, **params):
        """带缓存的服务器获取"""
        cache_key = f"registry:servers:{hash(frozenset(params.items()))}"
        
        # 尝试从缓存获取
        cached = self.redis.get(cache_key)
        if cached:
            return json.loads(cached)
        
        # 从 Registry 获取
        servers = self.fetch_servers(**params)
        
        # 缓存结果
        self.redis.setex(
            cache_key, 
            timedelta(seconds=self.cache_ttl),
            json.dumps(servers)
        )
        
        return servers

2. 错误处理 

import time
import random

class RobustRegistryClient:
    def __init__(self, max_retries=3, base_delay=1):
        self.max_retries = max_retries
        self.base_delay = base_delay
    
    def fetch_with_retry(self, url, **kwargs):
        """带重试的请求"""
        for attempt in range(self.max_retries):
            try:
                response = requests.get(url, timeout=30, **kwargs)
                response.raise_for_status()
                return response.json()
                
            except requests.RequestException as e:
                if attempt == self.max_retries - 1:
                    raise
                
                # 指数退避
                delay = self.base_delay * (2 ** attempt) + random.uniform(0, 1)
                time.sleep(delay)

3. 增量更新 

class IncrementalRegistrySync:
    def __init__(self, last_sync_file="last_sync.txt"):
        self.last_sync_file = last_sync_file
    
    def get_last_sync_time(self):
        """获取上次同步时间"""
        try:
            with open(self.last_sync_file, 'r') as f:
                return datetime.fromisoformat(f.read().strip())
        except FileNotFoundError:
            return datetime.min
    
    def save_sync_time(self, sync_time):
        """保存同步时间"""
        with open(self.last_sync_file, 'w') as f:
            f.write(sync_time.isoformat())
    
    def sync_changes(self):
        """同步变更（一旦 Registry 支持 updated_at 过滤）"""
        last_sync = self.get_last_sync_time()
        
        # 将来的 API 调用
        # servers = self.fetch_servers(updated_after=last_sync)
        
        # 目前获取所有服务器并在本地过滤
        all_servers = self.fetch_all_servers()
        
        self.save_sync_time(datetime.utcnow())
        return all_servers

下一步

查看完整的 API 规范 学习如何发布自己的服务器 了解 Registry 的技术架构 理解 Registry 的设计理念
建议: 在生产环境中使用子注册中心而不是直接使用官方 Registry，这样您可以添加缓存、过滤和增强功能。