跳到主要内容MCP 集成实战:连接 AI 与外部系统 | 极客日志PythonAI
MCP 集成实战:连接 AI 与外部系统
Model Context Protocol (MCP) 的概念及其在 Claude Code 中的应用。内容涵盖 MCP 的定义、三大能力(Tools、Resources、Prompts)以及与 Plugin 的区别。详细讲解了 MCP 配置文件的编写位置与格式,列举了官方常用服务器。重点演示了如何使用 FastMCP 框架开发自定义 Python 服务器,并通过 RAGFlow 知识库集成和 Jira 任务管理两个实际案例,展示了如何让 AI 访问内部文档和执行外部操作。最后提供了错误处理、安全考虑及性能优化等最佳实践建议。
beaabea3 浏览 引言:当 AI 助手遇到"信息孤岛"
使用 Claude Code 一段时间后,你可能遇到过这些场景:
场景 1:你想让 AI 帮忙查一下公司内部文档里关于某个 API 的说明,但它只能说"我无法访问你的内部系统"。
场景 2:你正在修复一个 Bug,想让 AI 看看 Jira 上的问题描述,却不得不手动复制粘贴一大段文字。
场景 3:你希望 AI 能直接查询数据库验证数据,而不是每次都让你手动执行 SQL 再把结果贴回来。
这些问题的本质是:Claude Code 是一个"信息孤岛",它只能访问本地文件系统,无法触及你的知识库、项目管理工具、数据库等外部系统。
MCP(Model Context Protocol) 就是解决这个问题的答案——它是 Anthropic 推出的开放协议,让 AI 能够安全、标准化地连接外部世界。
一、MCP 是什么?
1.1 一句话定义
MCP 是 AI 与外部系统之间的"USB 接口"。
就像 USB 让各种设备(键盘、鼠标、硬盘)能用统一的方式连接电脑,MCP 让各种外部服务(知识库、数据库、API)能用统一的方式连接 AI。
1.2 MCP 的三大能力
| 能力类型 | 说明 | 示例 |
|---|
| Tools(工具) | 让 AI 执行操作 | 搜索知识库、创建 Jira 任务、发送消息 |
| Resources(资源) | 向 AI 暴露数据 | 配置文件内容、数据库 Schema、API 文档 |
| Prompts(提示词) | 预定义的交互模板 | 代码审查模板、报告生成模板 |
实际开发中,Tools 是最常用的能力,本文的案例也主要围绕 Tools 展开。
1.3 MCP vs Plugin:有什么区别?
在第 10 篇我们讲过 Plugin,它和 MCP 有什么不同?
| 维度 | Plugin | MCP |
|---|
| 运行位置 | Claude Code 进程内 | 独立进程,通过协议通信 |
| 开发语言 | 必须是 TypeScript/JavaScript | 任意语言(Python、Go、Rust…) |
| 隔离性 | 共享 Claude Code 环境 | 完全隔离,更安全 |
| 适用场景 | 轻量级功能增强 | 连接外部系统、复杂集成 |
简单理解:Plugin 像是给 Claude Code'装插件',MCP 像是让 Claude Code'连外设'。
二、MCP 配置与使用
2.1 配置文件位置
Claude Code 的 MCP 配置文件位于:
~/.claude/mcp.json
项目目录/.mcp.json
2.2 配置文件格式
{
"mcpServers": {
"ragflow": {
"command": "python",
"args": ["/path/to/mcp_server.py", "--address", "ragflow.example.com", "--api-key", "your-api-key"]
},
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/allowed/dir"]
}
}
}
command:启动 MCP 服务器的命令
args:命令行参数
env:环境变量(可选,用于传递敏感信息)
2.3 常用官方 MCP 服务器
Anthropic 和社区提供了一些开箱即用的 MCP 服务器:
| 服务器 | 安装命令 | 用途 |
|---|
| filesystem | npx @modelcontextprotocol/server-filesystem | 访问指定目录的文件 |
| sqlite | npx @modelcontextprotocol/server-sqlite | 查询 SQLite 数据库 |
| fetch | npx @modelcontextprotocol/server-fetch | 发起 HTTP 请求 |
| github | npx @modelcontextprotocol/server-github | GitHub 操作 |
| brave-search | npx @modelcontextprotocol/server-brave-search | 网络搜索 |
2.4 验证 MCP 连接
配置完成后,在 Claude Code 中可以通过 /mcp 命令查看已连接的 MCP 服务器及其提供的工具。
三、自定义 MCP 服务器开发
当官方服务器无法满足需求时,就需要自己开发 MCP 服务器。Python 生态推荐使用 FastMCP 框架,它大大简化了开发流程。
3.1 FastMCP 框架
FastMCP 是 MCP 的 Python SDK 封装,提供了类似 Flask/FastAPI 的装饰器语法:
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("my_server")
@mcp.tool(name="hello", description="打招呼")
def hello(name: str) -> str:
return f"Hello, {name}!"
if __name__ == "__main__":
mcp.run(transport='stdio')
FastMCP("server_name"):创建服务器实例
@mcp.tool():装饰器定义工具,name 是工具名,description 是 AI 看到的说明
mcp.run(transport='stdio'):以 stdio 模式启动(Claude Code 使用此模式通信)
3.2 开发流程
- 安装依赖:
pip install mcp
- 创建服务器:使用 FastMCP 定义工具
- 本地测试:
python mcp_server.py --help
- 配置连接:编辑
~/.claude/mcp.json
- 验证集成:在 Claude Code 中测试工具调用
四、案例 1:RAGFlow 知识库集成
这是一个真实的生产案例,将 RAGFlow 知识库接入 Claude Code,让 AI 能够搜索和管理内部文档。
4.1 需求分析
- 搜索知识库中的相关内容
- 上传新文档到知识库
- 列出数据集和文档
4.2 核心实现
from mcp.server.fastmcp import FastMCP
from core.ragflow_api import get_datasets, retrieve_chunks, upload_document
import argparse
mcp = FastMCP("ragflow_mcp")
ADDRESS = None
API_KEY = None
@mcp.tool(name="search_chunks", description="从数据集中检索相关的文本块")
def search_chunks(question: str, dataset_id: str = None, page_size: int = 5) -> str:
"""
从知识库中搜索与问题相关的内容。
:param question: 要搜索的问题或关键词
:param dataset_id: 数据集 ID,不提供则搜索所有数据集
:param page_size: 返回结果数量,默认 5 条
"""
if dataset_id:
dataset_ids = [dataset_id]
else:
dataset_ids = get_all_dataset_ids()
result = retrieve_chunks(
address=ADDRESS,
api_key=API_KEY,
question=question,
dataset_ids=dataset_ids,
page_size=page_size,
similarity_threshold=0.1
)
if result and result.get('code') == 0:
chunks = result.get('data', {}).get('chunks', [])
if not chunks:
return f"未找到与 '{question}' 相关的内容"
formatted = []
for i, chunk in enumerate(chunks, 1):
content = chunk.get('content', '')
similarity = chunk.get('similarity', 0)
doc_name = chunk.get('document_keyword', '未知文档')
formatted.append(f"{i}. 【{doc_name}】(相似度:{similarity:.3f})\n{content[:500]}")
return f"找到 {len(chunks)} 个相关结果:\n\n" + "\n\n".join(formatted)
return f"检索失败:{result.get('message', '未知错误')}"
@mcp.tool(name="upload_documents", description="上传文档到知识库")
def upload_documents_tool(file_paths: list, dataset_id: str = None) -> str:
"""
上传本地文档到 RAGFlow 知识库。
:param file_paths: 文件路径列表
:param dataset_id: 目标数据集 ID
"""
if isinstance(file_paths, str):
file_paths = [file_paths]
results = []
for file_path in file_paths:
if not os.path.exists(file_path):
results.append({"file": file_path, "error": "文件不存在"})
continue
result = upload_document(
address=ADDRESS,
api_key=API_KEY,
dataset_id=dataset_id,
file_path=file_path,
document_name=os.path.basename(file_path)
)
results.append({"file": file_path, "result": result})
return json.dumps(results, ensure_ascii=False, indent=2)
def main():
global ADDRESS, API_KEY
parser = argparse.ArgumentParser(description='RAGFlow MCP Server')
parser.add_argument('--address', required=True, help='RAGFlow 服务器地址')
parser.add_argument('--api-key', required=True, help='API 密钥')
args = parser.parse_args()
ADDRESS = args.address
API_KEY = args.api_key
result = get_datasets(address=ADDRESS, api_key=API_KEY, page=1, page_size=1)
if not result or result.get('code') != 0:
print("❌ RAGFlow 连接失败", file=sys.stderr)
sys.exit(1)
print("✅ RAGFlow 连接成功", file=sys.stderr)
mcp.run(transport='stdio')
if __name__ == "__main__":
main()
4.3 配置与使用
{
"mcpServers": {
"ragflow": {
"command": "python",
"args": ["/path/to/mcp_server_stdio.py", "--address", "ragflow.your-company.com", "--api-key", "ragflow-xxxxxx"]
}
}
}
你:帮我在知识库里搜一下"用户认证流程"相关的文档
Claude Code:[调用 search_chunks 工具] 找到 3 个相关结果:
1. 【认证模块设计文档】(相似度:0.892) 用户认证采用 JWT + Refresh Token 双令牌机制...
2. 【API 接口规范】(相似度:0.756) POST /api/auth/login 用户登录接口...
五、案例 2:Jira 集成
让 Claude Code 能够直接操作 Jira,实现开发与任务管理的无缝衔接。
5.1 工具定义
from mcp.server.fastmcp import FastMCP
import requests
mcp = FastMCP("jira_mcp")
JIRA_URL = None
JIRA_TOKEN = None
@mcp.tool(name="get_issue", description="获取 Jira 任务详情")
def get_issue(issue_key: str) -> str:
"""
获取指定 Jira 任务的详细信息。
:param issue_key: 任务编号,如 PROJ-123
"""
response = requests.get(
f"{JIRA_URL}/rest/api/2/issue/{issue_key}",
headers={"Authorization": f"Bearer {JIRA_TOKEN}"},
timeout=10
)
if response.status_code == 200:
data = response.json()
fields = data.get('fields', {})
return f"""
任务:{issue_key}
标题:{fields.get('summary','')}
状态:{fields.get('status', {}).get('name', '')}
优先级:{fields.get('priority', {}).get('name', '')}
描述:{fields.get('description','')[:500]}
"""
return f"获取任务失败:{response.status_code}"
@mcp.tool(name="add_comment", description="给 Jira 任务添加评论")
def add_comment(issue_key: str, comment: str) -> str:
"""
给指定任务添加评论。
:param issue_key: 任务编号
:param comment: 评论内容
"""
response = requests.post(
f"{JIRA_URL}/rest/api/2/issue/{issue_key}/comment",
headers={"Authorization": f"Bearer {JIRA_TOKEN}", "Content-Type": "application/json"},
json={"body": comment},
timeout=10
)
if response.status_code == 201:
return f"✅ 评论已添加到 {issue_key}"
return f"❌ 添加评论失败:{response.status_code}"
@mcp.tool(name="update_status", description="更新 Jira 任务状态")
def update_status(issue_key: str, status: str) -> str:
"""
更新任务状态。
:param issue_key: 任务编号
:param status: 目标状态(如 "In Progress", "Done")
"""
trans_resp = requests.get(
f"{JIRA_URL}/rest/api/2/issue/{issue_key}/transitions",
headers={"Authorization": f"Bearer {JIRA_TOKEN}"}
)
if trans_resp.status_code != 200:
return f"获取状态转换失败"
transitions = trans_resp.json().get('transitions', [])
target = next((t for t in transitions if t['name'].lower() == status.lower()), None)
if not target:
available = [t['name'] for t in transitions]
return f"无效状态。可用状态:{available}"
response = requests.post(
f"{JIRA_URL}/rest/api/2/issue/{issue_key}/transitions",
headers={"Authorization": f"Bearer {JIRA_TOKEN}", "Content-Type": "application/json"},
json={"transition": {"id": target['id']}}
)
if response.status_code == 204:
return f"✅ {issue_key} 状态已更新为 {status}"
return f"❌ 状态更新失败"
5.2 实际使用场景
你:我已经修复了 PROJ-456 的 Bug,帮我更新一下 Jira 状态,并添加修复说明
Claude Code:好的,我来更新 Jira。
[调用 add_comment] ✅ 评论已添加到 PROJ-456
[调用 update_status] ✅ PROJ-456 状态已更新为 Done
已完成:
1. 添加了修复说明评论
2. 将任务状态更新为 Done
六、最佳实践
6.1 错误处理
MCP 工具的返回值会直接展示给 AI,因此错误信息要清晰易懂:
return str(e)
return f"❌ 操作失败:{str(e)}。请检查参数是否正确。"
6.2 安全考虑
- 敏感信息:API Key 等通过命令行参数或环境变量传入,不要硬编码
- 权限控制:MCP 服务器应该只暴露必要的操作,避免过度授权
- 输入校验:对用户输入进行校验,防止注入攻击
{
"mcpServers": {
"myserver": {
"command": "python",
"args": ["server.py"],
"env": {
"API_KEY": "${MYSERVER_API_KEY}"
}
}
}
}
6.3 性能优化
- 连接复用:使用连接池,避免每次请求都建立新连接
- 超时设置:所有外部调用都要设置合理的超时时间
- 结果缓存:对于不常变化的数据,可以在 MCP 服务器端缓存
总结
MCP 是 Claude Code 连接外部世界的桥梁:
- 官方 + 社区服务器:覆盖文件系统、数据库、HTTP 请求等常见场景
- FastMCP 框架:用 Python 快速开发自定义服务器
- 真实案例:RAGFlow 知识库集成让 AI 能搜索内部文档,Jira 集成实现开发流程自动化
通过 MCP,Claude Code 从一个只能操作本地文件的工具,进化为能够连接企业各种系统的"能力枢纽"。
微信扫一扫,关注极客日志
微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
相关免费在线工具
- RSA密钥对生成器
生成新的随机RSA私钥和公钥pem证书。 在线工具,RSA密钥对生成器在线工具,online
- Mermaid 预览与可视化编辑
基于 Mermaid.js 实时预览流程图、时序图等图表,支持源码编辑与即时渲染。 在线工具,Mermaid 预览与可视化编辑在线工具,online
- curl 转代码
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
- Base64 字符串编码/解码
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
- Base64 文件转换器
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online
- Markdown转HTML
将 Markdown(GFM)转为 HTML 片段,浏览器内 marked 解析;与 HTML转Markdown 互为补充。 在线工具,Markdown转HTML在线工具,online