技术

使用 MCP Server 管理博客文章

将博客编辑功能封装为 MCP Server，让 Claude Code 直接管理文章

李志奇 约 6 分钟 1,441 字

什么是 MCP

MCP (Model Context Protocol) 是 Anthropic 推出的协议，用于连接 AI 模型与外部工具和数据源。

MCP 传输模式

MCP 支持两种传输模式：

stdio 模式

┌─────────────────┐         ┌─────────────────┐
│   MCP Client    │  stdin  │   MCP Server    │
│  (Claude Code)  │ ◄─────► │  (Python 进程)   │
│                 │  stdout │                 │
└─────────────────┘         └─────────────────┘
  • MCP Client 启动 MCP Server 子进程
  • 通过标准输入输出通信
  • 适用于本地工具

SSE 模式

┌─────────────────┐   HTTP   ┌─────────────────┐
│   MCP Client    │ ◄──────► │   MCP Server    │
│    (Sage)       │   SSE    │   (远程服务)     │
│                 │          │                 │
│  POST 请求 ─────┼─────────►│  处理请求        │
│                 │          │                  │
│  ◄──────────────┼──────────│  SSE 推送响应    │
└─────────────────┘          └─────────────────┘

SSE 模式有两个通道：

通道 方向 用途
SSE 流 (GET /sse) Server → Client 推送事件和响应
POST 端点 (/messages/) Client → Server 发送 JSON-RPC 请求

连接流程：

Client                                    Server
  |                                          |
  |--- GET /sse (SSE 长连接) --------------->|
  |<-- SSE event: "endpoint" ----------------|  告知 POST 端点 URL
  |                                          |
  |--- POST /messages/ (JSON-RPC 请求) ----->|
  |<-- SSE event: "message" -----------------|  返回响应

两种模式对比

特性 stdio SSE
运行方式 本地子进程 远程 HTTP 服务
通信方式 stdin/stdout 管道 HTTP + SSE
部署位置 本地 可远程部署
多客户端 不支持 支持多客户端连接
防火墙 无需 需开放端口

MCP Server 开发

正确方式：使用 FastMCP 内置 SSE 支持

from mcp.server.fastmcp import FastMCP

# 初始化时配置 host 和 port
mcp = FastMCP("blog-editor", host="0.0.0.0", port=8765)

@mcp.tool()
async def list_articles(status: str = None) -> str:
    """列出博客文章列表"""
    ...

@mcp.tool()
async def create_article(slug: str, title: str, ...) -> str:
    """创建新文章"""
    ...

# SSE 模式运行（自动创建 Starlette 应用）
mcp.run(transport="sse")

关键点：

  1. host 和 port 在 FastMCP 初始化时设置，不是在 run() 方法
  2. run(transport="sse") 自动启动 Starlette 服务
  3. 默认端点：/sse 和 /messages/（注意 messages 有尾斜杠）

错误方式：手动挂载 SseServerTransport

# ❌ 错误：会导致 TypeError: 'NoneType' object is not callable
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route

sse = SseServerTransport("/messages")

async def handle_sse(request):
    async with sse.connect_sse(...):
        await mcp._mcp_server.run(...)
    # 没有 return Response，Starlette 无法处理！

async def handle_messages(request):
    await sse.handle_post_message(...)
    # 同样没有 return！

app = Starlette(routes=[
    Route("/sse", endpoint=handle_sse),
    Route("/messages", endpoint=handle_messages, methods=["POST"]),
])

问题原因：SseServerTransport.connect_sse() 和 handle_post_message() 是 ASGI 应用，不是请求处理函数。它们直接操作 scope/receive/send，不需要返回 Response。但 Starlette Route 的 endpoint 期望返回 Response 对象。

提供的 MCP 工具

工具 功能 参数
list_articles 列出文章列表 status: draft/published
get_article 获取文章详情 id: 文章ID
create_article 创建新文章 slug, title, content, tags, categories
update_article 更新文章 id + 可选字段
publish_article 发布文章 id
delete_article 删除文章 id

MCP Client 开发

Sage MCP Client 架构

Sage 作为 AI 助手平台，内置了 MCP Client 功能：

┌─────────────────────────────────────────────────────────┐
│                      Sage 架构                          │
├─────────────────────────────────────────────────────────┤
│  前端 (Vanilla JS)                                      │
│  ├── MCP 管理面板                                        │
│  ├── 传输类型选择 (stdio/SSE)                            │
│  └── 服务启停控制                                        │
├─────────────────────────────────────────────────────────┤
│  后端 (FastAPI)                                         │
│  ├── MCPServerConfig (配置模型)                          │
│  ├── MCPManager (进程/连接管理)                          │
│  └── MCP API (增删改查接口)                              │
└─────────────────────────────────────────────────────────┘

配置模型

# backend/agents/mcp/mcp_config.py
from pydantic import BaseModel
from typing import Literal, Optional

class MCPServerConfig(BaseModel):
    id: str
    name: str
    transport_type: Literal["stdio", "sse"] = "stdio"

    # stdio 模式配置
    command: str
    args: List[str] = []
    env_vars: Dict[str, str] = {}

    # sse 模式配置
    url: Optional[str] = None

MCP Manager 支持两种模式

# backend/agents/mcp/mcp_manager.py
from mcp.client.stdio import stdio_client
from mcp.client.sse import sse_client

async def start_server(self, mcp_id: str):
    config = await mcp_config_manager.get(mcp_id)
    exit_stack = AsyncExitStack()

    if config.transport_type == "sse":
        # SSE 模式：连接到远程 MCP 服务器
        transport_context = sse_client(config.url)
        transport = await exit_stack.enter_async_context(transport_context)
        read, write = transport
    else:
        # stdio 模式：启动本地进程
        server_params = StdioServerParameters(
            command=config.command,
            args=config.args,
        )
        transport_context = stdio_client(server_params)
        transport = await exit_stack.enter_async_context(transport_context)
        read, write = transport

    # 统一的 Session 处理
    session = await exit_stack.enter_async_context(ClientSession(read, write))
    await session.initialize()

    # 加载工具
    tools = await load_mcp_tools(session)

前端 MCP 管理界面

// frontend/js/sage-ui.js
async showAddMcpModal() {
    // 传输类型选择器
    const fields = [
        { name: 'name', label: '服务名称', required: true },
        { name: 'transport_type', label: '传输类型', type: 'radio',
          options: [
              { value: 'stdio', label: 'stdio (本地命令)' },
              { value: 'sse', label: 'SSE (远程服务)' }
          ]
        },
        // stdio 字段（条件显示）
        { name: 'command', label: '执行命令' },
        { name: 'args', label: '参数' },
        // sse 字段（条件显示）
        { name: 'url', label: 'SSE URL' }
    ];
}

版本兼容性注意事项

MCP 库版本需要保持一致：

# 检查版本
pip show mcp | grep Version

# 升级
pip install mcp --upgrade

如果 Server 和 Client 版本不一致，可能导致连接失败。

配置方式

方式一：stdio 模式（Claude Code）

在项目目录创建 .mcp.json：

{
  "mcpServers": {
    "blog-editor": {
      "command": "python",
      "args": ["backend/mcp_server.py"],
      "env": {
        "POSTGRES_HOST": "172.17.0.1"
      }
    }
  }
}

方式二：SSE 模式（Sage）

  1. 启动 MCP Server（SSE 模式）：

    POSTGRES_HOST=172.17.0.1 python mcp_server.py --mode sse

  2. 在 Sage Web 界面添加：

    • 服务名称：blog-editor
    • 传输类型：SSE
    • SSE URL：http://172.17.0.1:8765/sse

  3. 启动服务后，Sage 可调用博客工具

使用示例

Claude Code (stdio)

用户：列出所有草稿文章
Claude：调用 list_articles(status="draft")

用户：创建一篇关于 MCP 的文章
Claude：调用 create_article(slug="mcp-intro", title="MCP 介绍", ...)

用户：发布这篇文章
Claude：调用 publish_article(id=3)

Sage (SSE)

在 Sage 对话中：

用户：列出我的博客文章
Sage：[调用 blog-editor 的 list_articles 工具]
      目前有 5 篇已发布文章，2 篇草稿...

用户：帮我创建一篇新文章，标题是"我的新博客"
Sage：[调用 create_article 工具]
      已创建文章，slug 是 "my-new-blog"...

技术要点

  1. FastMCP 内置 SSE：使用 mcp.run(transport="sse")，不要手动挂载 SseServerTransport
  2. 配置方式：host/port 在 FastMCP 初始化时设置
  3. 端点路径：默认 /sse 和 /messages/（注意尾斜杠）
  4. 双传输支持：stdio 适合本地工具，SSE 适合远程服务
  5. 版本兼容：Server 和 Client 的 mcp 库版本需一致
  6. 统一 Session API：无论哪种传输模式，ClientSession 接口一致

总结

通过 MCP Server，博客管理变得非常简单：

  • stdio 模式：Claude Code 本地调用，无需网络
  • SSE 模式：Sage 远程连接，支持分布式部署
  • 使用 FastMCP 内置 SSE 支持，避免手动集成 Starlette
  • 统一的工具接口，AI 无需关心传输细节
  • 无缝集成到开发工作流中
标签

相关阅读

Hello from Sage

Sage 的自我介绍：一个集工具、记忆与 MCP 生态于一身的企业级智能助手。

2026/05/20

MCP 三种传输方式详解：stdio、SSE 与 Streamable HTTP

深入解析 MCP 协议的三种传输方式，重点介绍 2025 年新标准 Streamable HTTP——专为公网部署设计的传输方式，以及各方式的适用场景对比。

2026/05/20

Claude code 源码学习

2026/06/14