外观
一文搞懂企业内部 MCP Server 开发与注册
为什么企业需要自己的 MCP Tool Server
大模型接入企业内部系统时，最常见的做法是为每个智能体单独写函数调用：查工单、查库存、创建审批、检索知识库，每个平台都重复对接一遍。随着工具数量增加，权限、参数校验、日志审计和接口版本会迅速失控。
MCP（Model Context Protocol）的价值，是把“工具如何被发现、参数如何描述、结果如何返回”标准化。企业只需要实现一个内部 MCP Server，智能体、IDE 助手或工作流引擎即可通过统一协议发现 tools，并在授权后完成 tool-use。
本文用一个“员工差旅助手”场景，演示从零开发可运行的企业内部工具 Server，并注册到智能体工作流。示例使用 Python 标准库实现 JSON-RPC over stdio，避免初学者被框架版本差异卡住；生产环境可再替换为官方 SDK 或网关部署。
方案对比：函数调用、插件与 MCP Server
在动手前，需要明确三种集成方式的边界。
| 方案 | 接入成本 | 复用性 | 权限治理 | 适用场景 |
|---|---|---|---|---|
| 应用内 function calling | 低 | 弱 | 跟随业务服务 | 单一智能体、少量临时工具 |
| 传统插件/OpenAPI 包装 | 中 | 中 | 依赖网关 | 已有 REST API、调用方固定 |
| MCP Server | 中 | 高 | 可统一鉴权与审计 | 多智能体、多客户端、企业工具市场 |
MCP 不是让模型直接访问数据库，也不是把业务逻辑搬进提示词。它更像一个“工具适配器”：对上暴露结构化 tool schema，对下调用经过封装的内部 API，并负责鉴权、参数校验、错误归一化和审计日志。
MCP 工具调用链路与职责边界
下图展示一次典型的 tool-use：智能体先列出工具，再根据用户意图选择工具，MCP Server 校验身份和参数后访问内部系统，最后把结构化结果交还给模型生成回答。
需要强调的是，MCP Server 应保持“无业务决策”或“弱业务决策”。是否批准差旅、是否允许退款，仍应由后端系统根据制度判断；模型只负责理解意图、补全参数和解释结果，不能把最终授权交给大模型。
开发前准备：定义工具边界与安全模型
本示例实现三个工具：
travel_policy_search：查询差旅制度，只读、低风险。trip_budget_get：查询部门预算余额，需要员工身份和部门权限。trip_request_create：创建出差申请，属于写操作，需要二次确认和完整审计。
建议按风险分级，而不是把所有接口一次性暴露给智能体：
- 只读工具可自动调用，但必须限制返回字段，避免泄露薪资、身份证号等敏感信息。
- 写入工具必须使用幂等键，防止模型重试造成重复单据。
- 删除、审批、支付类动作不要直接开放给通用智能体，应进入人工确认流程。
从零实现一个可运行的 MCP Server
下面代码保存为 travel_mcp_server.py，通过 python travel_mcp_server.py 启动。它使用 stdin/stdout 传输 JSON-RPC，日志写入 stderr，避免污染协议消息。
python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""企业差旅工具 MCP Server（教学版，可直接通过 stdio 测试）"""
import sys
import json
import uuid
import traceback
from typing import Any, Dict
SERVER_NAME = "travel-tools"
SERVER_VERSION = "1.0.0"
# 生产环境应替换为真实内部 API；这里用内存数据演示
MOCK_POLICY = {
"hotel": "一线城市酒店标准不超过 500 元/晚，需保留发票。",
"flight": "高铁 5 小时以内优先选择二等座；超标准需直属主管审批。"
}
MOCK_BUDGET = {"platform": 12800.5, "sales": 3600.0}
MOCK_REQUESTS = []
def log(message: str) -> None:
# stdout 只能输出协议消息，诊断日志必须写 stderr
print(f"[mcp-log] {message}", file=stderr_safe(), flush=True)
def stderr_safe():
return sys.stderr
def tool_definitions() -> list[Dict[str, Any]]:
"""返回工具清单。description 会直接影响模型选择工具的准确率。"""
return [
{
"name": "travel_policy_search",
"description": "查询公司差旅制度。适用于询问酒店标准、交通工具、报销规则。",
"inputSchema": {
"type": "object",
"properties": {
"topic": {
"type": "string",
"enum": ["hotel", "flight"],
"description": "制度主题：hotel 表示酒店，flight 表示交通"
}
},
"required": ["topic"]
}
},
{
"name": "trip_budget_get",
"description": "查询当前员工所在部门的差旅预算余额。",
"inputSchema": {
"type": "object",
"properties": {
"department": {
"type": "string",
"enum": ["platform", "sales"],
"description": "部门标识"
}
},
"required": ["department"]
}
},
{
"name": "trip_request_create",
"description": "创建出差申请。仅在用户确认城市、日期和事由后调用。",
"inputSchema": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "出差城市"},
"start_date": {"type": "string", "description": "开始日期，格式 YYYY-MM-DD"},
"end_date": {"type": "string", "description": "结束日期，格式 YYYY-MM-DD"},
"reason": {"type": "string", "description": "出差事由"},
"idempotency_key": {
"type": "string",
"description": "幂等键，由客户端生成，防止重复创建"
}
},
"required": ["city", "start_date", "end_date", "reason", "idempotency_key"]
}
}
]
def call_tool(name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""执行工具并返回 MCP content。"""
if name == "travel_policy_search":
topic = arguments.get("topic")
if topic not in MOCK_POLICY:
raise ValueError("topic 只能是 hotel 或 flight")
data = {"topic": topic, "policy": MOCK_POLICY[topic]}
elif name == "trip_budget_get":
department = arguments.get("department")
if department not in MOCK_BUDGET:
raise PermissionError("无权查询该部门预算")
data = {"department": department, "budget_remaining": MOCK_BUDGET[department]}
elif name == "trip_request_create":
required = ["city", "start_date", "end_date", "reason", "idempotency_key"]
missing = [key for key in required if not arguments.get(key)]
if missing:
raise ValueError(f"缺少必填字段：{','.join(missing)}")
# 真实系统应根据 idempotency_key 去重
for item in MOCK_REQUESTS:
if item["idempotency_key"] == arguments["idempotency_key"]:
data = item
break
else:
data = {
"request_id": f"TR-{uuid.uuid4().hex[:8].upper()}",
"status": "pending_manager_approval",
**arguments
}
MOCK_REQUESTS.append(data)
else:
raise ValueError(f"未知工具：{name}")
return {
"content": [
{"type": "text", "text": json.dumps(data, ensure_ascii=False)}
],
"isError": False
}
def handle_request(request: Dict[str, Any]) -> Dict[str, Any] | None:
req_id = request.get("id")
method = request.get("method")
params = request.get("params", {})
# notification 没有 id，不需要响应
if req_id is None:
log(f"收到通知：{method}")
return None
if method == "initialize":
result = {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {"listChanged": False}},
"serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION}
}
elif method == "tools/list":
result = {"tools": tool_definitions()}
elif method == "tools/call":
try:
result = call_tool(params.get("name", ""), params.get("arguments", {}))
except Exception as exc:
log(traceback.format_exc())
result = {
"content": [{"type": "text", "text": f"工具调用失败：{exc}"}],
"isError": True
}
else:
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32601, "message": f"方法不存在：{method}"}
}
return {"jsonrpc": "2.0", "id": req_id, "result": result}
def main() -> None:
for line in sys.stdin:
line = line.strip()
if not line:
continue
try:
request = json.loads(line)
response = handle_request(request)
if response is not None:
print(json.dumps(response, ensure_ascii=False), flush=True)
except json.JSONDecodeError:
error = {
"jsonrpc": "2.0",
"id": None,
"error": {"code": -32700, "message": "JSON 解析失败"}
}
print(json.dumps(error, ensure_ascii=False), flush=True)
if __name__ == "__main__":
main()可以用下面的输入逐行测试，注意每条消息必须是一行 JSON：
text
{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}
{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}
{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"travel_policy_search","arguments":{"topic":"hotel"}}}注册到智能体：客户端配置与调用策略
支持 MCP 的智能体宿主通常允许配置一个 stdio 类型 Server。配置思路如下，不同平台字段名可能略有差异：
json
{
"mcpServers": {
"travel-tools": {
"command": "python",
"args": ["/opt/company/mcp/travel_mcp_server.py"],
"env": {
"EMPLOYEE_ID": "${EMPLOYEE_ID}",
"INTERNAL_API_TOKEN": "${INTERNAL_API_TOKEN}"
}
}
}
}注册后，智能体工作流一般经历四步：识别用户意图、拉取 tools/list、由模型生成 name + arguments、宿主调用 tools/call。对于写操作，建议在工作流中增加确认节点，而不是让模型立即执行。
可复用的系统提示词片段如下：
text
你是企业差旅助手。调用工具时必须遵守：
1. 涉及制度查询，优先使用 travel_policy_search，不要编造公司标准。
2. 创建出差申请前，必须向用户确认城市、开始日期、结束日期和事由。
3. 调用 trip_request_create 时必须生成唯一 idempotency_key。
4. 工具返回错误时，先解释缺失字段或权限问题，再向用户索要补充信息。
5. 不得猜测预算、审批结果或报销金额。生产化：从演示 Server 到企业服务
演示代码适合理解协议，真正上线还需要补齐四类能力。
第一是传输与部署。stdio 适合本地 IDE 和桌面助手；多人共享的智能体更适合 Streamable HTTP 或网关模式，由平台统一处理连接、扩缩容和版本路由。
第二是身份透传。不要让所有请求共用一个超级 Token。客户端应把员工身份、租户 ID、会话 ID 传入 Server，Server 再换发短期、最小权限的内部 API Token。
第三是数据最小化。工具返回给模型的内容应经过字段白名单过滤。例如预算接口只返回余额，不返回部门负责人手机号；工单接口只返回摘要，不返回客户合同全文。
第四是可观测性。每次 tool-use 至少记录请求 ID、员工 ID、工具名、参数摘要、结果状态、耗时和错误码。参数中的敏感字段要脱敏，避免审计日志变成新的数据泄露源。
避坑总结：MCP 落地最容易踩的 5 个坑
把调试信息打印到 stdout，导致协议解析失败。 现象是客户端提示 JSON 解析错误或连接立即断开。MCP stdio 通道中 stdout 只能返回协议消息，所有日志必须写 stderr，或接入统一日志系统。
工具描述过于笼统，模型频繁选错工具。 例如把描述写成“查询信息”，模型无法判断查制度还是查预算。修正方法是在
description中写清适用意图、典型问法和边界，并为枚举字段提供业务含义。参数 Schema 只写类型不写约束。 日期、金额、城市编码如果不限制格式，模型可能传“下周一”“大概三千”。应使用
format、enum、正则或服务端强校验，并在错误信息中返回可修复提示。写接口没有幂等设计，重试后重复创建单据。 常见现象是网络超时后模型再次调用，生成两张出差申请。所有创建类工具都必须要求
idempotency_key，后端按键去重，而不是依赖模型记住是否成功。把敏感数据和高权限动作一次性暴露给模型。 模型可能在总结时复述身份证号、薪资或客户信息。上线前应按工具分级、字段脱敏、权限校验和人工确认建立治理流程；删除、审批、支付不要做成自动 tool-use。
MCP 的核心收益不是“又多了一个协议”，而是让企业工具可以被安全、标准、可审计地复用。先从只读、低频、边界清晰的内部工具开始，跑通鉴权、Schema、审计和错误处理，再逐步接入写操作，才是智能体工作流落地最稳妥的路径。
免责声明
本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。