Day 5：MCP 服务化——让任何系统都能调你的分析能力

昨天的问题

前四天的代码跑得很好——但都只在你自己的电脑上。你写的 analyze_one_with_cache() 函数，你的同事用不了，你的前端用不了，你的老板更用不了。 天的问题前四天的代码跑得很好——但都只在你自己的电脑上。你写的 analyze_one_with_cache() 函数，你的同事用不了，你的前端用不了，你的老板更用不了。

"能不能打包成一个服务？" 当然可以。但问题来了——用什么协议？REST API？gRPC？直接 Python 函数调用？

答案是 MCP——Model Context Protocol。 项目一 Day 6 你已经见过它了。今天为你的文档分析功能做一个 MCP Server。

今天做什么

把前四天积累的三个核心能力——合同分析、文档摘要、文档分类——包装成 MCP 工具。写一个 MCP Server，定义三个工具，注册在 stdio 协议上。任何 MCP 客户端都能发现并调用它们。

你需要安装

pip install mcp

⚠️ 注意：MCP Python SDK 仍在快速迭代。本书代码基于 0.x 版本 API。如果安装后报 ImportError，请尝试 pip install mcp==0.9.4 锁定版本。

📁 代码目录

doc_analyzer/
├── sample_contract.txt
├── day1.py
├── day2.py
├── day3.py
├── day4.py
└── doc_analyzer_server.py     ← 新增！MCP Server，60行

代码

# doc_analyzer_server.py — MCP版本的文档分析服务

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import asyncio

# ===== 创建 MCP Server =====
server = Server("doc-analyzer")

# ===== 1. 注册工具列表 =====
@server.list_tools()
async def list_tools():
    """
    告诉 MCP 客户端："我能做这些事"。
    客户端拿到这个列表后展示给用户——用户选哪个，
    就调用下面 call_tool 里的对应分支。
    """
    return [
        Tool(
            name="analyze_contract",
            description="分析合同文档，提取甲方、乙方、合作内容、期限等关键信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {
                        "type": "string",
                        "description": "合同全文"
                    }
                },
                "required": ["text"]
            }
        ),
        Tool(
            name="summarize_document",
            description="对文档内容生成100字左右的摘要",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "文档全文"}
                },
                "required": ["text"]
            }
        ),
        Tool(
            name="classify_document",
            description="将文档分类为：合同 / 简历 / 报告 / 其他",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "文档全文"}
                },
                "required": ["text"]
            }
        ),
    ]

# ===== 2. 实现工具调用 =====
@server.call_tool()
async def call_tool(name: str, args: dict):
    """
    当客户端调用某个工具时，这个函数被触发。
    
    根据 name 决定执行哪个逻辑——就是前四天已经写完的分析能力，
    只不过现在包了一层 MCP 接口。
    """
    text = args.get("text", "")
    
    # --- 工具1：合同分析 ---
    if name == "analyze_contract":
        lines = [
            f"📋 合同分析报告",
            f"",
            f"文档长度：{len(text)} 字",
        ]
        
        # 从文本中提取关键信息（前四天练过的能力）
        if "甲方" in text:
            # 简单提取 —— 真实场景用 Day2 的 Pydantic + LLM
            for line in text.split("\n"):
                if "甲方" in line:
                    lines.append(f"甲方信息：{line.strip()}")
                    break
        if "乙方" in text:
            for line in text.split("\n"):
                if "乙方" in line:
                    lines.append(f"乙方信息：{line.strip()}")
                    break
        if "期限" in text or "有效期" in text:
            lines.append("⚠️  包含合同期限条款，请仔细核对日期")
        if "知识产权" in text:
            lines.append("⚠️  包含知识产权条款，建议重点审核")
        
        return [TextContent(type="text", text="\n".join(lines))]
    
    # --- 工具2：文档摘要 ---
    elif name == "summarize_document":
        # 简单摘要策略：取前100字 + 后20字
        if len(text) <= 100:
            summary = text
        else:
            opening = text[:100].replace("\n", " ")
            closing = text[-30:].replace("\n", " ")
            summary = f"{opening}... ...{closing}"
        
        return [TextContent(
            type="text",
            text=f"📄 文档摘要（{len(text)}字）：\n{summary}"
        )]
    
    # --- 工具3：文档分类 ---
    elif name == "classify_document":
        # 基于关键词的分类 —— 快速且实用
        scores = {
            "合同": 0,
            "简历": 0,
            "报告": 0,
        }
        
        if any(w in text for w in ["合同", "协议", "甲方", "乙方", "签署"]):
            scores["合同"] += 1
        if any(w in text for w in ["简历", "工作经历", "教育背景", "技能"]):
            scores["简历"] += 1
        if any(w in text for w in ["报告", "分析", "总结", "数据", "调研"]):
            scores["报告"] += 1
        
        top = max(scores, key=scores.get)
        category = top if scores[top] > 0 else "其他"
        
        detail = "\n".join([f"   {k}: {v} 分" for k, v in scores.items()])
        return [TextContent(
            type="text",
            text=f"📂 文档分类：{category}\n关键词得分：\n{detail}"
        )]
    
    return [TextContent(type="text", text="未知分析类型")]

# ===== 3. 启动服务 =====
async def main():
    """
    MCP Server 通过 stdio（标准输入输出）与客户端通信。
    这意味着：
    - 不需要开端口
    - 不需要配 CORS
    - 客户端用 subprocess 启动这个脚本就行
    """
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

测试

你的 MCP Server

MCP Server 不能直接 python doc_analyzer_server.py 跑——它等着客户端通过 stdio 跟它通信。测试方式有两种：

方式一：通过 MCP 客户端配置

{
  "mcpServers": {
    "doc-analyzer": {
      "command": "python",
      "args": ["doc_analyzer_server.py"],
      "cwd": "/path/to/doc_analyzer/"
    }
  }
}

方式二：写一个简单的测试脚本

# test_mcp.py —— 快速验证 MCP Server
import subprocess
import json

# 启动 MCP Server 子进程
proc = subprocess.Popen(
    ["python", "doc_analyzer_server.py"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# 发送 list_tools 请求（MCP 协议 JSON-RPC 格式）
request = json.dumps({
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/list",
    "params": {}
}) + "\n"

stdout, stderr = proc.communicate(input=request)
print("MCP Server 返回的工具列表：")
print(stdout)

你学到了什么

MCP 把"函数"变成了"工具"。 前四天你写的是 Python 函数——analyze_with_cache()、batch_analyze()。今天你把这些函数包装成了 MCP 工具——有名字、有描述、有输入格式。任何 MCP 客户端都能发现并调用它们。

跟项目一 MCP 的关系： 项目一 Day 6 的 MCP 是"客服工具"（查订单、查退货政策），今天是"文档分析工具"（合同分析、摘要、分类）。协议一样，工具不同。你会写一个 MCP Server，就会写所有 MCP Server——这就是标准化的威力。

生产价值： 你写的这个 MCP Server，可以被 Cursor、Claude Desktop、你同事写的内部平台调用。它不是"一个脚本"，而是"一个服务"。你掌握的不是怎么写一个别人用不了的东西，而是怎么写一个任何人都能通过标准协议调用的东西。

明天的预告

明天是项目二的最后一天。用 Streamlit 做一个 Web 界面——上传文档，调用 MCP Server，在浏览器里看分析结果。最后做项目的完整总结。

---

Day 5 完成。你的文档分析能力现在是 MCP 标准服务了——任何 MCP 客户端都能发现和调用。