MCP 协议集成：扩展 Hermes 的能力边界
TL;DR: 本教程深入讲解如何通过 Model Context Protocol (MCP) 将 Hermes Agent 与外部工具无缝集成。你将学习构建自定义 MCP 服务器、连接第三方 API、实现实时数据获取，以及构建一个集成 GitHub、Notion 和自定义数据源的智能工作流。完成本部分后，你的 Hermes Agent 将具备与任何外部系统交互的能力。
🎯 学习目标
完成本教程后，你将能够：
- 理解 MCP 协议的核心架构与通信模型
- 构建自定义 MCP 服务器（Python/TypeScript）
- 实现与 GitHub API、Notion API 的深度集成
- 创建支持实时数据获取的 Hermes 技能
- 调试和优化 MCP 连接性能
📋 前置知识回顾
在开始之前，确保你已掌握：
- Part 1-2: Hermes Agent 基本配置与多角色协作
- Part 3: Skill 系统的创建与管理（我们将基于此扩展）
- Part 4: Cron 任务自动化（可选，但有助于理解集成场景）
如果你需要复习，建议先阅读 [Part 3: Skill 系统深度解析] 中关于 hermes skill create 和 hermes skill test 的部分。
🔧 MCP 协议基础
什么是 MCP？
Model Context Protocol (MCP) 是 Anthropic 于 2024 年提出的开放协议，旨在标准化 AI 模型与外部工具的通信方式。与传统的 REST API 不同，MCP 专门为 LLM 交互设计，支持：
- 动态工具发现：Agent 运行时自动发现可用工具
- 结构化上下文：工具返回的数据自动注入模型上下文
- 双向通信：支持流式响应和异步回调
- 安全沙箱：工具执行在隔离环境中运行
Hermes Agent 中的 MCP 架构
┌─────────────────────────────────────────┐
│ Hermes Agent │
│ ┌──────────────────────────────────┐ │
│ │ MCP Client Layer │ │
│ │ (hermes mcp connect/discover) │ │
│ └──────────┬───────────────────────┘ │
│ │ │
│ ┌──────────▼───────────────────────┐ │
│ │ Skill Execution Engine │ │
│ │ (hermes skill run --mcp) │ │
│ └──────────┬───────────────────────┘ │
└─────────────┼───────────────────────────┘
│
┌─────────▼─────────┐
│ MCP Server(s) │
│ (Python/TS/Go) │
└─────────┬─────────┘
│
┌─────────▼─────────┐
│ External APIs │
│ (GitHub/Notion/DB)│
└───────────────────┘
🚀 实战：构建你的第一个 MCP 服务器
环境准备
# 确保 Hermes Agent 版本 >= 2.5.0
hermes --version
# 应输出: Hermes Agent v2.5.0 (MCP support enabled)
# 安装 MCP SDK
pip install mcp-python-sdk==0.8.0
# 验证安装
python -c "from mcp import Server; print('MCP SDK ready')"
项目结构
mcp-servers/
├── github-server/
│ ├── server.py
│ ├── requirements.txt
│ └── config.yaml
├── notion-server/
│ ├── server.py
│ ├── requirements.txt
│ └── config.yaml
└── custom-server/
├── server.py
├── data_processor.py
└── config.yaml
案例 1：GitHub 数据获取服务器
这是最实用的集成场景之一。我们将构建一个 MCP 服务器，让 Hermes Agent 能实时获取 GitHub 仓库的统计数据和 Issue 信息。
server.py:
# github-server/server.py
from mcp import Server, Tool, ToolContext
from typing import Dict, List, Optional
import httpx
import os
from datetime import datetime, timedelta
class GitHubMCPServer(Server):
def __init__(self):
super().__init__(
name="github-data-server",
version="1.0.0",
description="Real-time GitHub repository data access"
)
self.github_token = os.getenv("GITHUB_TOKEN")
self.base_url = "https://api.github.com"
async def register_tools(self):
"""注册所有可用工具"""
# 工具 1: 获取仓库统计
self.register_tool(Tool(
name="get_repo_stats",
description="获取指定仓库的详细统计信息，包括星标、Fork、Issue 等",
parameters={
"owner": {"type": "string", "description": "仓库所有者"},
"repo": {"type": "string", "description": "仓库名称"},
"include_contributors": {
"type": "boolean",
"description": "是否包含贡献者统计",
"default": False
}
},
handler=self.handle_get_repo_stats
))
# 工具 2: 搜索 Issue
self.register_tool(Tool(
name="search_issues",
description="搜索仓库中的 Issue 和 Pull Request",
parameters={
"owner": {"type": "string"},
"repo": {"type": "string"},
"state": {
"type": "string",
"enum": ["open", "closed", "all"],
"default": "open"
},
"labels": {
"type": "array",
"items": {"type": "string"},
"description": "按标签过滤"
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 50,
"default": 10
}
},
handler=self.handle_search_issues
))
# 工具 3: 获取活跃开发者
self.register_tool(Tool(
name="get_active_developers",
description="获取最近 30 天最活跃的贡献者",
parameters={
"owner": {"type": "string"},
"repo": {"type": "string"},
"days": {
"type": "integer",
"default": 30
}
},
handler=self.handle_get_active_developers
))
async def handle_get_repo_stats(self, params: Dict, context: ToolContext) -> Dict:
"""处理仓库统计请求"""
owner = params["owner"]
repo = params["repo"]
headers = {"Authorization": f"token {self.github_token}"}
async with httpx.AsyncClient() as client:
# 并行获取多个数据源
repo_resp, issues_resp, prs_resp = await asyncio.gather(
client.get(f"{self.base_url}/repos/{owner}/{repo}", headers=headers),
client.get(f"{self.base_url}/repos/{owner}/{repo}/issues?state=open", headers=headers),
client.get(f"{self.base_url}/repos/{owner}/{repo}/pulls?state=open", headers=headers)
)
if repo_resp.status_code != 200:
return {"error": f"Repository not found: {repo_resp.status_code}"}
repo_data = repo_resp.json()
result = {
"repository": f"{owner}/{repo}",
"stars": repo_data["stargazers_count"],
"forks": repo_data["forks_count"],
"open_issues": repo_data["open_issues_count"],
"open_pull_requests": len(prs_resp.json()),
"last_updated": repo_data["updated_at"],
"primary_language": repo_data.get("language"),
"topics": repo_data.get("topics", []),
"license": repo_data.get("license", {}).get("spdx_id", "N/A")
}
# 可选：获取贡献者统计
if params.get("include_contributors"):
contrib_resp = await client.get(
f"{self.base_url}/repos/{owner}/{repo}/contributors?per_page=5",
headers=headers
)
if contrib_resp.status_code == 200:
result["top_contributors"] = [
{"login": c["login"], "contributions": c["contributions"]}
for c in contrib_resp.json()
]
return result
async def handle_search_issues(self, params: Dict, context: ToolContext) -> Dict:
"""搜索 Issue 和 PR"""
owner = params["owner"]
repo = params["repo"]
state = params.get("state", "open")
labels = params.get("labels", [])
limit = params.get("limit", 10)
headers = {"Authorization": f"token {self.github_token}"}
# 构建查询参数
query_parts = [f"repo:{owner}/{repo}", f"state:{state}"]
if labels:
for label in labels:
query_parts.append(f"label:{label}")
query = " ".join(query_parts)
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{self.base_url}/search/issues",
headers=headers,
params={"q": query, "per_page": limit, "sort": "updated"}
)
if resp.status_code != 200:
return {"error": f"Search failed: {resp.status_code}"}
data = resp.json()
issues = []
for item in data["items"][:limit]:
issues.append({
"number": item["number"],
"title": item["title"],
"state": item["state"],
"type": "pull_request" if "pull_request" in item else "issue",
"labels": [l["name"] for l in item["labels"]],
"created_at": item["created_at"],
"updated_at": item["updated_at"],
"author": item["user"]["login"],
"url": item["html_url"]
})
return {
"total_count": data["total_count"],
"issues": issues,
"query": query
}
async def handle_get_active_developers(self, params: Dict, context: ToolContext) -> Dict:
"""获取活跃开发者"""
owner = params["owner"]
repo = params["repo"]
days = params.get("days", 30)
since_date = (datetime.now() - timedelta(days=days)).isoformat()
headers = {"Authorization": f"token {self.github_token}"}
async with httpx.AsyncClient() as client:
# 获取最近的 commits
resp = await client.get(
f"{self.base_url}/repos/{owner}/{repo}/commits",
headers=headers,
params={"since": since_date, "per_page": 100}
)
if resp.status_code != 200:
return {"error": f"Failed to get commits: {resp.status_code}"}
commits = resp.json()
# 统计每个开发者的提交数
developer_stats = {}
for commit in commits:
author = commit.get("author")
if author and author.get("login"):
login = author["login"]
if login not in developer_stats:
developer_stats[login] = {
"commits": 0,
"avatar": author.get("avatar_url", ""),
"last_commit": commit["commit"]["author"]["date"]
}
developer_stats[login]["commits"] += 1
developer_stats[login]["last_commit"] = commit["commit"]["author"]["date"]
# 按提交数排序
sorted_developers = sorted(
developer_stats.items(),
key=lambda x: x[1]["commits"],
reverse=True
)
return {
"repository": f"{owner}/{repo}",
"period_days": days,
"total_commits": len(commits),
"active_developers": [
{
"login": login,
"commits": stats["commits"],
"last_commit": stats["last_commit"]
}
for login, stats in sorted_developers[:10]
]
}
# 启动服务器
if __name__ == "__main__":
server = GitHubMCPServer()
server.run(host="0.0.0.0", port=8000)
config.yaml:
# github-server/config.yaml
server:
name: github-data-server
version: "1.0.0"
port: 8000
rate_limit: 30 # 每分钟请求数
authentication:
type: token
env_var: GITHUB_TOKEN
logging:
level: INFO
format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
案例 2：Notion 数据库集成
这个服务器允许 Hermes Agent 读写 Notion 数据库，实现内容管理和知识库集成。
server.py:
# notion-server/server.py
from mcp import Server, Tool, ToolContext
from typing import Dict, List, Optional
import httpx
import os
class NotionMCPServer(Server):
def __init__(self):
super().__init__(
name="notion-database-server",
version="1.0.0",
description="Notion database read/write operations"
)
self.notion_token = os.getenv("NOTION_TOKEN")
self.base_url = "https://api.notion.com/v1"
self.headers = {
"Authorization": f"Bearer {self.notion_token}",
"Notion-Version": "2022-06-28",
"Content-Type": "application/json"
}
async def register_tools(self):
# 工具 1: 查询数据库
self.register_tool(Tool(
name="query_database",
description="查询 Notion 数据库中的条目",
parameters={
"database_id": {"type": "string", "description": "数据库 ID"},
"filter": {
"type": "object",
"description": "过滤条件（Notion 过滤语法）",
"optional": True
},
"sort": {
"type": "array",
"description": "排序规则",
"optional": True
},
"page_size": {
"type": "integer",
"default": 10,
"maximum": 100
}
},
handler=self.handle_query_database
))
# 工具 2: 创建页面
self.register_tool(Tool(
name="create_database_page",
description="在 Notion 数据库中创建新条目",
parameters={
"database_id": {"type": "string"},
"properties": {
"type": "object",
"description": "页面属性（字段名: 值）"
},
"content": {
"type": "object",
"description": "页面内容（富文本块）",
"optional": True
}
},
handler=self.handle_create_page
))
# 工具 3: 更新页面
self.register_tool(Tool(
name="update_page",
description="更新 Notion 页面属性",
parameters={
"page_id": {"type": "string"},
"properties": {"type": "object"},
"archived": {
"type": "boolean",
"description": "是否归档页面",
"optional": True
}
},
handler=self.handle_update_page
))
async def handle_query_database(self, params: Dict, context: ToolContext) -> Dict:
"""查询 Notion 数据库"""
database_id = params["database_id"]
page_size = params.get("page_size", 10)
request_body = {"page_size": page_size}
if "filter" in params:
request_body["filter"] = params["filter"]
if "sort" in params:
request_body["sorts"] = params["sort"]
async with httpx.AsyncClient() as client:
resp = await client.post(
f"{self.base_url}/databases/{database_id}/query",
headers=self.headers,
json=request_body
)
if resp.status_code != 200:
return {"error": f"Query failed: {resp.status_code}", "details": resp.text}
data = resp.json()
# 转换结果为可读格式
results = []
for page in data["results"]:
properties = {}
for prop_name, prop_data in page["properties"].items():
properties[prop_name] = self._extract_property_value(prop_data)
results.append({
"id": page["id"],
"url": page["url"],
"created_time": page["created_time"],
"last_edited_time": page["last_edited_time"],
"properties": properties
})
return {
"total_results": len(results),
"has_more": data.get("has_more", False),
"results": results
}
def _extract_property_value(self, prop_data: Dict) -> any:
"""提取 Notion 属性值"""
prop_type = prop_data.get("type", "")
if prop_type == "title":
return prop_data["title"][0]["plain_text"] if prop_data["title"] else ""
elif prop_type == "rich_text":
return prop_data["rich_text"][0]["plain_text"] if prop_data["rich_text"] else ""
elif prop_type == "number":
return prop_data.get("number")
elif prop_type == "select":
return prop_data["select"]["name"] if prop_data.get("select") else None
elif prop_type == "multi_select":
return [s["name"] for s in prop_data.get("multi_select", [])]
elif prop_type == "date":
return prop_data.get("date")
elif prop_type == "checkbox":
return prop_data.get("checkbox", False)
elif prop_type == "url":
return prop_data.get("url", "")
elif prop_type == "email":
return prop_data.get("email", "")
elif prop_type == "phone_number":
return prop_data.get("phone_number", "")
else:
return str(prop_data)
# 启动
if __name__ == "__main__":
server = NotionMCPServer()
server.run(host="0.0.0.0", port=8001)
🔗 集成到 Hermes Agent
配置 MCP 连接
# ~/.hermes/config.yaml
mcp:
servers:
github:
url: "http://localhost:8000"
timeout: 30
retry_count: 3
auth_token: "${GITHUB_MCP_TOKEN}" # 可选，用于 MCP 服务器间认证
notion:
url: "http://localhost:8001"
timeout: 15
retry_count: 2
custom:
url: "http://localhost:8002"
timeout: 60
health_check_interval: 30 # 秒
# 全局设置
global:
max_parallel_requests: 5
cache_ttl: 300 # 缓存工具结果 5 分钟
enable_discovery: true # 自动发现新工具
创建集成技能
现在我们将创建一个 Hermes Skill，利用上述 MCP 服务器实现智能代码审查助手。
hermes skill create code-review-assistant
skill.yaml:
# skills/code-review-assistant/skill.yaml
name: code-review-assistant
version: "1.0.0"
description: |
智能代码审查助手，结合 GitHub 数据和 Notion 知识库
自动分析 PR，检查代码规范，并记录审查结果到 Notion
mcp_servers:
- github
- notion
triggers:
- type: webhook
event: github.pull_request.opened
condition: "repository.topics contains 'review-required'"
- type: command
pattern: "review pr {owner}/{repo}#{pr_number}"
- type: cron
schedule: "0 9 * * 1-5" # 工作日早上 9 点
task: "check_pending_reviews"
workflow:
steps:
- name: fetch_pr_details
tool: github.get_repo_stats
params:
owner: "{owner}"
repo: "{repo}"
output: repo_stats
- name: analyze_pr
tool: github.search_issues
params:
owner: "{owner}"
repo: "{repo}"
state: open
labels: ["needs-review"]
output: pending_prs
- name: check_code_standards
tool: notion.query_database
params:
database_id: "${NOTION_CODE_STANDARDS_DB}"
filter:
property: "Language"
select:
equals: "{repo.primary_language}"
output: standards
- name: generate_review
prompt: |
基于以下信息进行代码审查：
仓库统计：{repo_stats}
待审查 PR：{pending_prs}
代码规范：{standards}
请生成一个详细的审查报告，包括：
1. 代码质量评估
2. 潜在问题
3. 改进建议
4. 是否通过审查
输出格式：JSON
output: review_result
- name: save_to_notion
tool: notion.create_database_page
params:
database_id: "${NOTION_REVIEWS_DB}"
properties:
Title:
title: "Review: {owner}/{repo}#{pr_number}"
Repository:
rich_text: "{owner}/{repo}"
PR Number:
number: "{pr_number}"
Status:
select: "{review_result.passed}"
Review Date:
date: "{current_date}"
content:
- type: heading_2
heading_2:
rich_text: [{text: {content: "审查报告"}}]
- type: code
code:
language: json
rich_text: [{text: {content: "{review_result.json}"}}]
dependencies:
- name: httpx
version: ">=0.27.0"
- name: mcp-python-sdk
version: ">=0.8.0"
测试集成
# 1. 启动 MCP 服务器
cd mcp-servers/github-server
GITHUB_TOKEN=ghp_xxx python server.py &
# 输出: [INFO] Starting GitHub MCP server on port 8000
cd ../notion-server
NOTION_TOKEN=secret_xxx python server.py &
# 输出: [INFO] Starting Notion MCP server on port 8001
# 2. 验证连接
hermes mcp connect --server github --url http://localhost:8000
# 输出: ✓ Connected to github-data-server v1.0.0
# 发现工具: get_repo_stats, search_issues, get_active_developers
hermes mcp connect --server notion --url http://localhost:8001
# 输出: ✓ Connected to notion-database-server v1.0.0
# 发现工具: query_database, create_database_page, update_page
# 3. 测试工具调用
hermes mcp call \
--server github \
--tool get_repo_stats \
--params '{"owner": "facebook", "repo": "react", "include_contributors": true}'
# 输出示例:
# {
# "repository": "facebook/react",
# "stars": 220000,
# "forks": 45000,
# "open_issues": 1200,
# "open_pull_requests": 85,
# "primary_language": "JavaScript",
# "top_contributors": [
# {"login": "gaearon", "contributions": 1500},
# {"login": "sophiebits", "contributions": 800}
# ]
# }
# 4. 运行集成技能
hermes skill run code-review-assistant \
--params '{"owner": "my-org", "repo": "my-project", "pr_number": 42}'
⚡ 高级用法：构建自定义数据处理服务器
对于需要复杂数据处理或实时信息获取的场景，我们可以构建一个更强大的自定义 MCP 服务器。
custom-server/server.py:
# custom-server/server.py
from mcp import Server, Tool, ToolContext
from typing import Dict, List, Optional
import httpx
import json
from datetime import datetime
import aiohttp
from bs4 import BeautifulSoup
import re
class DataProcessingServer(Server):
def __init__(self):
super().__init__(
name="data-processing-server",
version="2.0.0",
description="高级数据处理和实时信息获取"
)
async def register_tools(self):
# 工具 1: 实时网站数据提取
self.register_tool(Tool(
name="extract_website_data",
description="从任意网站提取结构化数据（支持 JavaScript 渲染）",
parameters={
"url": {"type": "string", "description": "目标 URL"},
"selectors": {
"type": "object",
"description": "CSS 选择器映射，如 {title: 'h1', price: '.price'}",
"optional": True
},
"use_headless": {
"type": "boolean",
"description": "是否使用无头浏览器渲染 JS",
"default": False
}
},
handler=self.handle_extract_website
))
# 工具 2: 技术栈分析
self.register_tool(Tool(
name="analyze_tech_stack",
description="分析网站的技术栈（框架、库、CDN 等）",
parameters={
"url": {"type": "string"},
"deep_scan": {
"type": "boolean",
"description": "是否深度扫描子页面",
"default": False
}
},
handler=self.handle_tech_analysis
))
# 工具 3: 代码质量分析
self.register_tool(Tool(
name="analyze_code_quality",
description="分析 GitHub 仓库的代码质量指标",
parameters={
"owner": {"type": "string"},
"repo": {"type": "string"},
"metrics": {
"
---
*Have questions? Join our [Discord community](https://discord.gg/smartotics) or follow us on [X](https://x.com/smartotics).*