第 41 章

Hermes 作为 MCP Server 对外提供服务

第41章 Hermes 作为 MCP Server 对外提供服务

导语

大多数人把 Hermes Agent 用作"消费者"——调用外部工具、读取数据库、操纵文件系统。但还有一种颠覆性用法：让 Hermes 本身成为 MCP Server，向 Claude Code、Cursor 等外部客户端暴露能力。这种"反向 MCP 架构"让企业能够把私有 Hermes 实例作为一等公民接入整个 AI 工具链，实现能力的复用与隔离。

本章将深入讲解反向 MCP 架构的设计原理，提供完整的 Python 实现，并分析它在哪些场景下是最优选择、在哪些场景下需要换思路。

41.1 为什么要把 Hermes 包装成 MCP Server

传统用法的局限

在标准 Hermes Agent 部署中，Hermes 是 MCP 协议的客户端：它连接到 filesystem、database、web-search 等 MCP Server，调用它们的工具来完成任务。这个模型简洁高效，但存在几个局限：

Claude Code 无法直接调用 Hermes 的推理能力
企业内部私有知识库的访问逻辑被锁在 Hermes 内部
多个 AI 客户端需要各自重复实现相同的业务逻辑

反向 MCP 的价值

把 Hermes 包装成 MCP Server 后，你获得了：

传统架构	反向 MCP 架构
Claude Code → 直接调用工具	Claude Code → Hermes MCP → 工具
每个客户端各自实现业务逻辑	业务逻辑集中在 Hermes
无法利用 Hermes 的长上下文能力	可借助 Hermes 64K+ 上下文做深度分析
无法复用 Hermes 的 Agent 工作流	Hermes 工作流作为工具暴露

41.2 反向 MCP 架构设计

架构图

┌─────────────────────────────────────────────────────────┐
│                    MCP 客户端层                           │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────┐  │
│  │ Claude Code │  │   Cursor    │  │  自定义 IDE 插件  │  │
│  └──────┬──────┘  └──────┬──────┘  └────────┬────────┘  │
└─────────┼────────────────┼──────────────────┼───────────┘
          │ MCP Protocol   │                  │
          ▼                ▼                  ▼
┌─────────────────────────────────────────────────────────┐
│              Hermes MCP Wrapper (本章实现)                │
│                                                         │
│  ┌──────────────────────────────────────────────────┐   │
│  │              MCP Server (JSON-RPC 2.0)           │   │
│  │  tools/list  tools/call  resources/read          │   │
│  └──────────────────────┬───────────────────────────┘   │
│                         │                               │
│  ┌──────────────────────▼───────────────────────────┐   │
│  │           Hermes Agent Core                      │   │
│  │  • 任务规划  • 工具选择  • 多步推理               │   │
│  └──────────────────────┬───────────────────────────┘   │
│                         │                               │
│  ┌──────────────────────▼───────────────────────────┐   │
│  │         Ollama / vLLM / llama.cpp                │   │
│  │              (Hermes 4 70B 推理引擎)              │   │
│  └──────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────┘
          │
          ▼ 调用下游工具（可选）
┌─────────────────────────────────────────────────────────┐
│  filesystem MCP │ database MCP │ web-search MCP │ ...   │
└─────────────────────────────────────────────────────────┘

核心设计原则

1. 工具粒度控制

不要把 Hermes 的所有能力塞进一个 run_agent 工具。根据业务域拆分成多个语义清晰的工具：

analyze_code(code, language, focus) → 代码分析
research_topic(query, depth)        → 深度研究
review_document(content, criteria)  → 文档评审
execute_workflow(name, params)      → 执行预定义工作流

2. 流式响应支持

Hermes 推理耗时可能达数十秒，必须实现 SSE（Server-Sent Events）或 MCP streaming，避免客户端超时。

3. 会话隔离

每个 MCP 客户端连接维护独立的 Hermes Agent 实例（或上下文 ID），防止跨客户端的上下文污染。

41.3 实现 Hermes MCP Wrapper（完整 Python 代码）

依赖安装

pip install mcp httpx asyncio pydantic python-dotenv
# mcp 是 Anthropic 官方 Python SDK for Model Context Protocol

项目结构

hermes-mcp-server/
├── server.py          # MCP Server 主入口
├── hermes_client.py   # 与 Hermes/Ollama 通信
├── tools.py           # 工具定义
├── config.py          # 配置管理
├── .env               # 环境变量
└── requirements.txt

config.py — 配置管理

# config.py
import os
from dataclasses import dataclass
from dotenv import load_dotenv

load_dotenv()

@dataclass
class HermesConfig:
    # Hermes 后端地址（Ollama/vLLM/llama.cpp）
    hermes_base_url: str = os.getenv("HERMES_BASE_URL", "http://localhost:11434")
    hermes_model: str = os.getenv("HERMES_MODEL", "nous-hermes2:70b-q4_0")
    
    # MCP Server 配置
    mcp_host: str = os.getenv("MCP_HOST", "127.0.0.1")
    mcp_port: int = int(os.getenv("MCP_PORT", "8765"))
    
    # Agent 行为配置
    max_tokens: int = int(os.getenv("MAX_TOKENS", "4096"))
    temperature: float = float(os.getenv("TEMPERATURE", "0.1"))
    context_window: int = int(os.getenv("CONTEXT_WINDOW", "65536"))
    
    # 超时配置（秒）
    request_timeout: int = int(os.getenv("REQUEST_TIMEOUT", "120"))
    stream_timeout: int = int(os.getenv("STREAM_TIMEOUT", "300"))

config = HermesConfig()

hermes_client.py — Hermes 通信层

# hermes_client.py
import httpx
import json
import asyncio
from typing import AsyncGenerator, Optional
from config import config

class HermesClient:
    """
    封装与 Hermes 推理后端的通信。
    支持 Ollama API 格式（也兼容 OpenAI API 格式的 vLLM）。
    """
    
    def __init__(self):
        self.client = httpx.AsyncClient(
            base_url=config.hermes_base_url,
            timeout=httpx.Timeout(config.request_timeout)
        )
    
    async def chat_completion(
        self,
        messages: list[dict],
        stream: bool = False,
        tools: Optional[list] = None
    ) -> dict | AsyncGenerator:
        """
        调用 Hermes 聊天接口。
        Ollama 使用 /api/chat，vLLM 使用 /v1/chat/completions。
        """
        # 检测后端类型
        if "11434" in config.hermes_base_url or "ollama" in config.hermes_base_url.lower():
            return await self._ollama_chat(messages, stream, tools)
        else:
            return await self._openai_chat(messages, stream, tools)
    
    async def _ollama_chat(
        self, 
        messages: list[dict],
        stream: bool,
        tools: Optional[list]
    ) -> dict | AsyncGenerator:
        """Ollama API 调用"""
        payload = {
            "model": config.hermes_model,
            "messages": messages,
            "stream": stream,
            "options": {
                "temperature": config.temperature,
                "num_predict": config.max_tokens,
                "num_ctx": config.context_window,
            }
        }
        if tools:
            payload["tools"] = tools
        
        if stream:
            return self._stream_ollama(payload)
        
        async with self.client as c:
            response = await c.post("/api/chat", json=payload)
            response.raise_for_status()
            return response.json()
    
    async def _stream_ollama(self, payload: dict) -> AsyncGenerator[str, None]:
        """流式输出 Ollama 响应"""
        async with httpx.AsyncClient(
            base_url=config.hermes_base_url,
            timeout=httpx.Timeout(config.stream_timeout)
        ) as client:
            async with client.stream("POST", "/api/chat", json=payload) as response:
                async for line in response.aiter_lines():
                    if line:
                        data = json.loads(line)
                        if not data.get("done", False):
                            content = data.get("message", {}).get("content", "")
                            if content:
                                yield content
    
    async def _openai_chat(
        self,
        messages: list[dict],
        stream: bool,
        tools: Optional[list]
    ) -> dict | AsyncGenerator:
        """OpenAI 兼容 API（vLLM 使用）"""
        payload = {
            "model": config.hermes_model,
            "messages": messages,
            "stream": stream,
            "temperature": config.temperature,
            "max_tokens": config.max_tokens,
        }
        if tools:
            payload["tools"] = tools
            payload["tool_choice"] = "auto"
        
        async with httpx.AsyncClient(
            base_url=config.hermes_base_url,
            timeout=httpx.Timeout(config.request_timeout)
        ) as client:
            if stream:
                return self._stream_openai(client, payload)
            response = await client.post("/v1/chat/completions", json=payload)
            response.raise_for_status()
            return response.json()
    
    async def _stream_openai(self, client, payload):
        async with client.stream("POST", "/v1/chat/completions", json=payload) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    data_str = line[6:]
                    if data_str == "[DONE]":
                        break
                    data = json.loads(data_str)
                    delta = data["choices"][0].get("delta", {})
                    content = delta.get("content", "")
                    if content:
                        yield content

hermes_client = HermesClient()

tools.py — 工具定义

# tools.py
"""
定义暴露给 MCP 客户端的工具。
每个工具对应一个 Hermes Agent 能力。
"""

HERMES_TOOLS = [
    {
        "name": "analyze_code",
        "description": (
            "使用 Hermes AI 深度分析代码。可以识别 bug、安全漏洞、"
            "性能问题、代码质量问题，并给出重构建议。"
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "code": {
                    "type": "string",
                    "description": "要分析的代码内容"
                },
                "language": {
                    "type": "string",
                    "description": "编程语言（python/javascript/rust/go等）",
                    "default": "auto"
                },
                "focus": {
                    "type": "string",
                    "enum": ["bugs", "security", "performance", "quality", "all"],
                    "description": "分析重点",
                    "default": "all"
                },
                "context": {
                    "type": "string",
                    "description": "额外上下文（如项目背景、业务逻辑说明）",
                    "default": ""
                }
            },
            "required": ["code"]
        }
    },
    {
        "name": "research_topic",
        "description": (
            "让 Hermes 对一个主题进行深度研究和分析。"
            "适合需要综合多方面信息、生成结构化报告的场景。"
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "研究主题或问题"
                },
                "depth": {
                    "type": "string",
                    "enum": ["quick", "standard", "deep"],
                    "description": "研究深度",
                    "default": "standard"
                },
                "output_format": {
                    "type": "string",
                    "enum": ["markdown", "json", "plain"],
                    "default": "markdown"
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "review_document",
        "description": "让 Hermes 对文档进行专业评审，包括内容质量、逻辑结构、表达清晰度等。",
        "inputSchema": {
            "type": "object",
            "properties": {
                "content": {
                    "type": "string",
                    "description": "文档内容"
                },
                "criteria": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "评审标准列表",
                    "default": ["accuracy", "clarity", "completeness", "structure"]
                },
                "tone": {
                    "type": "string",
                    "enum": ["strict", "balanced", "encouraging"],
                    "default": "balanced"
                }
            },
            "required": ["content"]
        }
    },
    {
        "name": "execute_workflow",
        "description": "执行预定义的 Hermes Agent 工作流。工作流封装了复杂的多步骤任务。",
        "inputSchema": {
            "type": "object",
            "properties": {
                "workflow_name": {
                    "type": "string",
                    "description": "工作流名称（如：code_review_pipeline、data_analysis、report_generation）"
                },
                "params": {
                    "type": "object",
                    "description": "工作流参数（键值对）"
                }
            },
            "required": ["workflow_name", "params"]
        }
    }
]

server.py — MCP Server 主入口

# server.py
import asyncio
import json
import logging
from typing import Any
import mcp.server.stdio
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp import types
from tools import HERMES_TOOLS
from hermes_client import hermes_client
from config import config

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("hermes-mcp")

# 创建 MCP Server 实例
app = Server("hermes-agent-mcp")

# ─── 工具列表 ────────────────────────────────────────────────────────────────

@app.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """返回所有可用工具的描述"""
    return [
        types.Tool(
            name=tool["name"],
            description=tool["description"],
            inputSchema=tool["inputSchema"]
        )
        for tool in HERMES_TOOLS
    ]

# ─── 工具调用 ────────────────────────────────────────────────────────────────

@app.call_tool()
async def handle_call_tool(
    name: str,
    arguments: dict[str, Any]
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    """处理来自 MCP 客户端的工具调用"""
    
    logger.info(f"Tool called: {name}, args: {list(arguments.keys())}")
    
    try:
        if name == "analyze_code":
            result = await _analyze_code(**arguments)
        elif name == "research_topic":
            result = await _research_topic(**arguments)
        elif name == "review_document":
            result = await _review_document(**arguments)
        elif name == "execute_workflow":
            result = await _execute_workflow(**arguments)
        else:
            result = f"错误：未知工具 '{name}'"
        
        return [types.TextContent(type="text", text=result)]
    
    except Exception as e:
        logger.error(f"Tool {name} failed: {e}")
        return [types.TextContent(
            type="text",
            text=f"工具执行失败: {str(e)}"
        )]

# ─── 工具实现 ────────────────────────────────────────────────────────────────

async def _analyze_code(
    code: str,
    language: str = "auto",
    focus: str = "all",
    context: str = ""
) -> str:
    focus_prompts = {
        "bugs": "重点识别逻辑错误、边界条件问题、潜在的运行时错误",
        "security": "重点识别安全漏洞：注入攻击、不安全的反序列化、敏感数据暴露等",
        "performance": "重点识别性能瓶颈：O(n²)算法、无效数据库查询、内存泄漏",
        "quality": "重点评估代码质量：可读性、可维护性、SOLID原则遵循度",
        "all": "全面分析：bug、安全、性能、代码质量"
    }
    
    system_prompt = """你是一位资深软件工程师，具备深度代码审查能力。
    分析时请结构化输出：
    1. 问题摘要（严重度：高/中/低）
    2. 具体问题列表（含行号建议）
    3. 修复建议（含代码示例）
    4. 总体评分（1-10）"""
    
    user_prompt = f"""请分析以下{language}代码：
    
分析重点：{focus_prompts.get(focus, focus_prompts['all'])}
{f'额外上下文：{context}' if context else ''}

```{language}
{code}

请提供详细的分析报告。"""

messages = [
    {"role": "system", "content": system_prompt},
    {"role": "user", "content": user_prompt}
]

response = await hermes_client.chat_completion(messages)
return response["message"]["content"]

async def _research_topic( query: str, depth: str = "standard", output_format: str = "markdown" ) -> str: depth_config = { "quick": "简要分析，300字以内，只列关键点", "standard": "标准深度，1000字左右，涵盖主要方面", "deep": "深度研究，2000字以上，包含细节分析、多角度对比、具体案例" }

messages = [
    {
        "role": "system",
        "content": f"你是一位专业研究员。请以{output_format}格式输出研究报告。"
    },
    {
        "role": "user",
        "content": f"研究主题：{query}\n\n深度要求：{depth_config[depth]}"
    }
]

response = await hermes_client.chat_completion(messages)
return response["message"]["content"]

async def _review_document( content: str, criteria: list = None, tone: str = "balanced" ) -> str: if criteria is None: criteria = ["accuracy", "clarity", "completeness", "structure"]

criteria_str = "、".join(criteria)
tone_desc = {
    "strict": "严格专业，直接指出所有问题",
    "balanced": "平衡客观，指出问题同时肯定优点",
    "encouraging": "鼓励为主，温和指出改进空间"
}

messages = [
    {
        "role": "system",
        "content": f"你是一位专业文档评审专家。评审风格：{tone_desc[tone]}"
    },
    {
        "role": "user",
        "content": f"请按照以下标准评审文档：{criteria_str}\n\n文档内容：\n{content}"
    }
]

response = await hermes_client.chat_completion(messages)
return response["message"]["content"]

async def _execute_workflow(workflow_name: str, params: dict) -> str: """执行预定义工作流""" workflow_prompts = { "code_review_pipeline": _build_code_review_prompt, "data_analysis": _build_data_analysis_prompt, "report_generation": _build_report_prompt, }

if workflow_name not in workflow_prompts:
    available = ", ".join(workflow_prompts.keys())
    return f"未知工作流 '{workflow_name}'。可用工作流：{available}"

prompt_builder = workflow_prompts[workflow_name]
messages = prompt_builder(params)

response = await hermes_client.chat_completion(messages)
return response["message"]["content"]

def _build_code_review_prompt(params: dict) -> list: return [ {"role": "system", "content": "你是 code review 专家，执行完整的代码评审流水线。"}, {"role": "user", "content": f"执行代码评审流水线，参数：{json.dumps(params, ensure_ascii=False)}"} ]

def _build_data_analysis_prompt(params: dict) -> list: return [ {"role": "system", "content": "你是数据分析专家，执行数据分析任务。"}, {"role": "user", "content": f"分析以下数据，参数：{json.dumps(params, ensure_ascii=False)}"} ]

def _build_report_prompt(params: dict) -> list: return [ {"role": "system", "content": "你是专业报告撰写专家。"}, {"role": "user", "content": f"生成报告，参数：{json.dumps(params, ensure_ascii=False)}"} ]

─── 启动 Server ─────────────────────────────────────────────────────────────

async def main(): logger.info(f"Hermes MCP Server 启动中...") logger.info(f"后端地址: {config.hermes_base_url}") logger.info(f"模型: {config.hermes_model}")

async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
    await app.run(
        read_stream,
        write_stream,
        InitializationOptions(
            server_name="hermes-agent-mcp",
            server_version="1.0.0",
            capabilities=app.get_capabilities(
                notification_options=None,
                experimental_capabilities={}
            )
        )
    )

if name == "main": asyncio.run(main())


---

## 41.4 配置示例

### Claude Code 配置（.mcp.json）

在项目根目录或 `~/.claude/` 下创建 `.mcp.json`：

```json
{
  "mcpServers": {
    "hermes-agent": {
      "command": "python",
      "args": ["/path/to/hermes-mcp-server/server.py"],
      "env": {
        "HERMES_BASE_URL": "http://localhost:11434",
        "HERMES_MODEL": "nous-hermes2:70b-q4_0",
        "MAX_TOKENS": "4096",
        "TEMPERATURE": "0.1",
        "CONTEXT_WINDOW": "65536",
        "REQUEST_TIMEOUT": "120"
      }
    }
  }
}

Cursor 配置（settings.json）

{
  "mcp": {
    "servers": {
      "hermes": {
        "command": "python",
        "args": ["/path/to/hermes-mcp-server/server.py"],
        "env": {
          "HERMES_BASE_URL": "http://localhost:11434",
          "HERMES_MODEL": "nous-hermes2:70b-q4_0"
        }
      }
    }
  }
}

环境变量文件（.env）

# Hermes 后端
HERMES_BASE_URL=http://localhost:11434
HERMES_MODEL=nous-hermes2:70b-q4_0

# 如果使用 vLLM（OpenAI 兼容接口）
# HERMES_BASE_URL=http://localhost:8000
# HERMES_MODEL=NousResearch/Hermes-4-70B

# 性能参数
MAX_TOKENS=4096
TEMPERATURE=0.1
CONTEXT_WINDOW=65536

# 超时（秒）
REQUEST_TIMEOUT=120
STREAM_TIMEOUT=300

Docker 一键启动

# docker-compose.yml（简化版，完整版见第47章）
version: "3.9"
services:
  hermes-mcp:
    build: .
    ports:
      - "8765:8765"
    environment:
      - HERMES_BASE_URL=http://ollama:11434
      - HERMES_MODEL=nous-hermes2:70b-q4_0
    depends_on:
      - ollama
  
  ollama:
    image: ollama/ollama:latest
    volumes:
      - ollama_data:/root/.ollama
    deploy:
      resources:
        reservations:
          devices:
            - capabilities: [gpu]

volumes:
  ollama_data:

41.5 使用场景与限制分析

场景	优势	示例
私有代码库智能分析	Hermes 可访问内网代码，Claude Code 通过 MCP 获得结果	安全审计、技术债分析
复杂业务规则推理	把 Hermes 上的领域知识作为服务暴露	合规检查、风险评估
长文档深度处理	利用 Hermes 64K 上下文处理超长文档	合同分析、技术文档总结
标准化工作流执行	封装重复的多步 Agent 工作流	代码 review 流水线
离线/私有部署	Hermes 本地运行，数据不出内网	金融、医疗、政府场景

已知限制

性能延迟

Claude Code → MCP 调用 → Hermes 推理 → 返回
     网络延迟 +    解析开销 +  推理延迟（5-60s）

Hermes 70B 单次推理需要 5-60 秒，对于需要即时响应的场景体验较差。

并发限制

单 GPU 实例通常只能串行处理请求
多客户端并发调用时需要排队或横向扩展（见第48章）

上下文管理复杂性

MCP 协议本身无状态，跨调用的上下文需要调用方管理
复杂的多轮对话场景需要在工具参数中传递历史

不适合的场景

需要 <1 秒响应的实时应用
高并发（>50 QPS）场景（需配合 vLLM + K8s）
简单查询（用直接 API 调用更高效）

本章小结

本章实现了完整的 Hermes MCP Wrapper，让 Hermes 从 MCP 消费者变成 MCP 提供者：

架构设计：清晰的分层架构，MCP Server → Hermes Agent Core → 推理引擎
工具粒度：按业务域拆分工具，而非暴露单一的通用接口
多后端支持：自动检测 Ollama/vLLM，切换无需修改代码
配置灵活性：通过环境变量控制所有关键参数

这种反向 MCP 架构的核心价值是能力复用：同一份 Hermes 部署，可以同时为 Claude Code、Cursor、自定义工具等多个客户端提供服务。

思考题

如果需要支持 Hermes 的流式输出（streaming），MCP 协议层需要做哪些修改？目前 MCP 规范对 streaming 的支持程度如何？
当多个 Claude Code 实例同时调用同一个 Hermes MCP Server 时，如何防止上下文混乱？设计一个会话隔离方案。
反向 MCP 架构中，Hermes 既是 MCP Server（对外），又需要调用下游 MCP Server（对内）。请画出完整的请求链路，并分析可能的循环依赖风险。

本章评分

4.9 / 5 (3 评分)