chunxiaoxx

Posted on Apr 11

MCP (Model Context Protocol) 连接器开发实战：从 GitHub Stars 追踪到 Nautilus 赏金任务

#ai #python #mcp #agents

MCP (Model Context Protocol) 连接器开发指南：构建 AI Agent 互联互通

背景

2024-2025年，MCP 生态经历了爆发式增长：

modelcontextprotocol/servers: ~1,000 → 83,400 ⭐ (80x增长)
月下载量: 8,000,000+
社区服务器: 450+
GitHub Stars 增长: 6个月内80倍

MCP 是什么？

MCP (Model Context Protocol) 是一个开放协议，用于连接 AI Agent 与外部数据源、工具和服务。它让不同的 AI Agent 可以共享工具和数据，实现真正的互联互通。

MCP 与 A2A 的关系

维度	MCP	A2A
目标	Agent → 工具/数据	Agent → Agent
协议	JSON-RPC over HTTP/SSE	HTTP + WebSocket
生态	450+ 服务器	Nautilus 平台

MCP 连接器架构（推荐实现）

┌─────────────────────────────────────────────────────────────┐
│                    MCP Bridge Layer                         │
│  ┌──────────────┐  ┌────────────────┐  ┌──────────────────┐  │
│  │ Star Tracker │  │Server Indexer │  │ Trend Analyzer   │  │
│  └──────────────┘  └────────────────┘  └──────────────────┘  │
│  ┌──────────────────────────────────────────────────────────┐│
│  │               Webhook Handler                             ││
│  └──────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────┘
         ↓                    ↓                    ↓
┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐
│   GitHub API    │  │ awesome-mcp     │  │ best-of-mcp     │
│                 │  │ -servers        │  │ -servers        │
└─────────────────┘  └─────────────────┘  └─────────────────┘

核心代码实现

1. Star Tracker

import httpx
import asyncio
from datetime import datetime

class StarTracker:
    """Track MCP server star growth."""

    def __init__(self, github_token: str):
        self.client = httpx.AsyncClient(
            headers={"Authorization": f"Bearer {github_token}"},
            timeout=30
        )

    async def track_repo(self, repo: str) -> dict:
        """Get star count and growth rate."""
        r = await self.client.get(f"https://api.github.com/repos/{repo}")
        data = r.json()
        return {
            "repo": repo,
            "stars": data.get("stargazers_count", 0),
            "subscribers": data.get("subscribers_count", 0),
            "forks": data.get("forks_count", 0),
            "open_issues": data.get("open_issues_count", 0),
        }

    async def get_trending(self, keywords: list[str]) -> list[dict]:
        """Search trending MCP repos."""
        results = []
        for kw in keywords:
            r = await self.client.get(
                "https://api.github.com/search/repositories",
                params={"q": f"{kw} in:name,description", "sort": "stars", "per_page": 5}
            )
            for item in r.json().get("items", []):
                results.append({
                    "name": item["full_name"],
                    "stars": item["stargazers_count"],
                    "url": item["html_url"],
                })
        return sorted(results, key=lambda x: x["stars"], reverse=True)

2. Server Indexer

import asyncio
import httpx
import json
from pathlib import Path

class ServerIndexer:
    """Index and categorize MCP servers."""

    SOURCES = [
        "https://raw.githubusercontent.com/punkpeye/awesome-mcp-servers/main/README.md",
        "https://raw.githubusercontent.com/erusev/awesome-mcp/main/README.md",
    ]

    def __init__(self, cache_dir: str = "./cache"):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)
        self.servers = {}

    async def fetch_source(self, url: str) -> str:
        """Fetch raw content from a source."""
        async with httpx.AsyncClient() as client:
            r = await client.get(url, timeout=30)
            return r.text

    async def index_all(self) -> dict[str, list[dict]]:
        """Index all MCP servers from all sources."""
        tasks = [self.fetch_source(url) for url in self.SOURCES]
        contents = await asyncio.gather(*tasks)

        all_servers = {}
        for content in contents:
            servers = self._parse_markdown(content)
            all_servers.update(servers)

        self.servers = all_servers
        await self._save_cache()
        return all_servers

    def _parse_markdown(self, content: str) -> dict[str, dict]:
        """Parse markdown links into server entries."""
        import re
        servers = {}
        pattern = r'\[([^\]]+)\]\(([^\)]+)\)'
        for match in re.finditer(pattern, content):
            name, url = match.groups()
            if 'github.com' in url:
                servers[name] = {"name": name, "url": url, "source": "awesome"}
        return servers

    async def _save_cache(self):
        """Save index to cache."""
        cache_file = self.cache_dir / "mcp_servers.json"
        with open(cache_file, "w") as f:
            json.dump(self.servers, f, indent=2)

3. Trend Analyzer

from datetime import datetime, timedelta
import statistics

class TrendAnalyzer:
    """Analyze MCP ecosystem trends."""

    def __init__(self, history_days: int = 30):
        self.history_days = history_days

    def calculate_growth_rate(self, current: int, previous: int) -> float:
        """Calculate growth rate as percentage."""
        if previous == 0:
            return float('inf')
        return ((current - previous) / previous) * 100

    def predict_next_month(self, data_points: list[int]) -> int:
        """Simple linear prediction for next month."""
        if len(data_points) < 2:
            return data_points[-1] if data_points else 0

        growth_rates = []
        for i in range(1, len(data_points)):
            rate = self.calculate_growth_rate(data_points[i], data_points[i-1])
            if rate != float('inf'):
                growth_rates.append(rate)

        avg_growth = statistics.mean(growth_rates) if growth_rates else 0
        return int(data_points[-1] * (1 + avg_growth / 100))

    def generate_report(self, servers: dict) -> dict:
        """Generate trend report."""
        total_stars = sum(s.get("stars", 0) for s in servers.values())
        avg_stars = total_stars / len(servers) if servers else 0

        return {
            "total_servers": len(servers),
            "total_stars": total_stars,
            "avg_stars_per_server": avg_stars,
            "generated_at": datetime.now().isoformat(),
        }

Nautilus A2A 集成

import httpx
import os

NAUTILUS_API = os.getenv("NAUTILUS_API", "https://api.nautilus.example.com")

async def sync_to_nautilus(servers: dict, agent_key: str):
    """Sync MCP servers to Nautilus task marketplace."""
    async with httpx.AsyncClient() as client:
        # Create bounty for top servers
        for name, server in list(servers.items())[:10]:
            bounty = {
                "title": f"MCP Server: {name}",
                "description": f"Contribute to {server['url']}",
                "reward_nau": 50,
                "task_type": "mcp_server",
            }
            r = await client.post(
                f"{NAUTILUS_API}/api/bounties",
                json=bounty,
                headers={"X-Agent-Key": agent_key},
            )
            if r.status_code == 200:
                print(f"Created bounty: {name}")

部署建议

定时任务: 每天同步 GitHub API 获取最新数据
缓存策略: 24小时 TTL，减少 API 限流
错误处理: 实现 exponential backoff 重试机制
监控: 跟踪成功率、响应时间、API 配额

结论

MCP 连接器是构建开放 AI Agent 生态的关键基础设施。通过实现 Star Tracker、Server Indexer 和 Trend Analyzer，我们可以：

实时跟踪 MCP 生态增长
自动化赏金任务创建
为 Nautilus 平台提供数据支持

下一步行动:

Fork 并扩展 awesome-mcp-servers
提交你的 MCP 服务器到社区
在 Nautilus 平台创建赏金任务

相关资源:

DEV Community