搜狐架构演进技术实践：我们如何用 MCP Registry 根治 AI Agent 的“千具之灾”

Python# tool_synchronizer.pyimport requestsimport scheduleimport timeimport jsonimport loggingfrom sentence_transformers import SentenceTransformerimport chromadb
# --- Configuration ---REGISTRY_API_ENDPOINT = "http://localhost:8080/v0/servers"CHROMA_DB_PATH = "./tooldb_prod"EMBEDDING_MODEL = "bge-large-zh-v1.5" # 强大的中英双语Embedding模型SYNC_INTERVAL_MINUTES = 5
# --- Setup ---logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")client = chromadb.PersistentClient(path=CHROMA_DB_PATH)model = SentenceTransformer(EMBEDDING_MODEL)collection = client.get_or_create_collection(name="mcp_tools_prod")
def fetch_tools_from_registry():    """从MCP Registry获取所有服务定义。"""    try:        response = requests.get(REGISTRY_API_ENDPOINT, timeout=10)        response.raise_for_status()        return response.json()    except requests.RequestException as e:        logging.error(f"从Registry获取失败: {e}")        return None
def process_and_index_tools(servers):    """处理工具元数据，并将其向量化索引，实现幂等同步。"""    if not servers:        logging.warning("无服务可处理。")        return
    active_tools = {s["meta"]["uuid"]: s for s in servers if s.get("status") == "active"}    logging.info(f"发现 {len(active_tools)} 个活跃工具待处理。")        # --- 1. 准备Upsert数据 ---    ids_to_upsert = list(active_tools.keys())    if not ids_to_upsert:         logging.info("没有活跃的工具需要更新或插入。")    else:        documents_to_upsert = []        metadatas_to_upsert = []
        for uuid, tool in active_tools.items():            # Pro-Tip: Embedding的文本质量直接决定检索效果。            # 融合工具名、描述和关键参数，能极大提升语义表达的丰富度。            param_names = ""            # Safely access parameters which might not exist            if "packages" in tool and tool["packages"]:                params = tool["packages"][0].get("parameters", [])                if params:                    param_names = ", ".join([p.get("name", "") for p in params])
            searchable_text = f"工具名: {tool.get("name", "")}; 功能描述: {tool.get("description", "")}; 参数: {param_names}"            documents_to_upsert.append(searchable_text)            metadatas_to_upsert.append({                "name": tool.get("name", ""),                "version": tool.get("version", ""),                "schema_json": json.dumps(tool) # 存储完整定义，供后续LLM使用            })                # --- 2. 执行幂等的Upsert操作 ---        collection.upsert(            ids=ids_to_upsert,            documents=documents_to_upsert,            metadatas=metadatas_to_upsert        )        logging.info(f"成功Upsert {len(ids_to_upsert)} 个工具到向量库。")        # --- 3. 处理删除项，保持数据一致性 ---    existing_ids_in_db = set(collection.get(include=[])["ids"])    active_ids_from_registry = set(active_tools.keys())    ids_to_delete = list(existing_ids_in_db - active_ids_from_registry)    if ids_to_delete:        collection.delete(ids=ids_to_delete)        logging.info(f"从向量库中删除 {len(ids_to_delete)} 个过时工具。")
def sync_job():    """同步任务的主函数。"""    logging.info("开始同步任务...")    servers_data = fetch_tools_from_registry()    if servers_data and "servers" in servers_data:        process_and_index_tools(servers_data["servers"])    logging.info("同步任务完成。")
if __name__ == "__main__":    sync_job() # 启动时立即执行一次    schedule.every(SYNC_INTERVAL_MINUTES).minutes.do(sync_job)    logging.info(f"调度器已启动，每 {SYNC_INTERVAL_MINUTES} 分钟同步一次。")    while True:        schedule.run_pending()        time.sleep(1)

Python# Part of agent_runtime.pyimport chromadbfrom sentence_transformers import SentenceTransformerimport json
class ToolRouter:    def __init__(self, chroma_client, embedding_model, top_k=3):        self.client = chroma_client        self.collection = self.client.get_collection(name="mcp_tools_prod")        self.model = embedding_model        self.top_k = top_k
    def route(self, user_query: str) -> list[dict]:        """根据用户查询，从向量库中检索最相关的K个工具的完整Schema。"""        # Ensure embeddings are normalized for cosine similarity searches        query_embedding = self.model.encode(user_query, normalize_embeddings=True).tolist()                results = self.collection.query(            query_embeddings=[query_embedding],            n_results=self.top_k,        )                if not results["metadatas"] or not results["metadatas"][0]:            return []                    retrieved_tools_schemas = [json.loads(meta["schema_json"]) for meta in results["metadatas"][0]]        return retrieved_tools_schemas
# --- 在Agent中的集成示例 ---# 假设已有main_llm客户端，如OpenAI()# from openai import OpenAI# main_llm = OpenAI(api_key="...")
def execute_agent_task(user_query: str, tool_router: ToolRouter):    # 1. 路由（Route）：动态筛选工具，上下文窗口永远轻量    print(f"Routing for query: "{user_query}"")    relevant_tool_schemas = tool_router.route(user_query)    if not relevant_tool_schemas:        print("No relevant tools found.")        # Handle case with no tools, maybe a direct LLM call without tools        return        print(f"Found {len(relevant_tool_schemas)} relevant tools: {[tool["name"] for tool in relevant_tool_schemas]}")
    # 2. 推理（Reason）：将筛选后的工具注入主LLM进行决策    # response = main_llm.chat.completions.create(    #     model="claude-3-opus-20240229", # Or GPT-4o    #     messages=[{"role": "user", "content": user_query}],    #     tools=relevant_tool_schemas,    #     tool_choice={"type": "auto"}    # )        # 3. 执行（Execute）：通过安全执行器调用LLM选择的工具    # ...后续的tool_calls处理逻辑...    print("Agent would now reason with the main LLM using the filtered tools.")