动手学agent应用开发笔记_task04_MCP的实现
本文介绍了MCP(Model Calling Protocol)协议的由来与实践。MCP基于OpenAI的Function Calling功能,允许大模型不仅生成文本还能执行任务。文章分析了Function Calling的开发痛点,如业务复杂性增加、代码维护困难等,并对比了无MCP协议(高耦合)与有MCP协议(灵活解耦)的架构差异。实践部分展示了如何使用fastmcp框架快速开发MCP服务器,包
学习连接:https://www.datawhale.cn/learn/content/220/5017
一、MCP的由来
Function Calling基础概念
- OpenAl接口协议下的一环。
- 允许大模型调用工具,不再只是文字的生产者,也能成为任务的执行者。
- Al Agent 的基础。
Function Calling案例
- 定义大模型可用的工具函数get_weather
- 通过tools字段注入上下文,告诉大模tools型"你可以使用的工具以及工具的描述"。
- 通过[{“role”:“user”“content”: “xxx”}]启动对话/任务。
- 通过大模型返回的tool_calls字段获取“大模型想要使用的工具数组"。
from openai import OpenAI
client = OpenAI(
api_key="<your api key>",
base_url="https://api.deepseek.com",
)
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather of a location, the user should supply a location first.",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
}
},
"required": ["location"]
},
}
},
]
def send_messages(messages):
response = client.chat.completions.create(
model="deepseek-chat",
messages=messages,
tools=tools
)
return response.choices[0].message
messages = [{"role": "user", "content": "How's the weather in Hangzhou?"}]
message = send_messages(messages)
print(f"User>\t {messages[0]['content']}")
tool = message.tool_calls[0]
messages.append(message)
messages.append({"role": "tool", "tool_call_id": tool.id, "content": "24℃"})
message = send_messages(messages)
print(f"Model>\t {message.content}")
Function Calling开发痛点分析
- 业务规模扩大带来的复杂性
- 代码维护性下降
- 系统不确定性挑战
- 开发心智负担加重
MCP协议
无MCP协议:耦合度高,开发和维护成本大
有MCP协议:后端服务与Agent解耦,扩展灵活
二、MCP实践
开发配置准备
- VS Code
- OpenMCP插件(超推荐!!!)https://openmcp.kirigaya.cn/
- Python 3.10+
- 大模型的API Token
技术选型
fastmcp(开发MCP服务器) + OpenMCP(调试验证MCP服务器)
fastmcp安装
pip install uv
uv add mcp "mcp[cli]"
uv的优点,便于管理,云服务友好
fastmcp 和 OpenMCP快速入门
fastmcp
# server.py
from mcp.server.fastmcp import FastMCP
# 创建一个 MCP Server
mcp = FastMCP("HelloMCP")
# 定义一个 prompt
@mcp.prompt()
def hello(name: str) -> str:
"""
hello prompt
"""
return f"hello {name}"
# 定义一个工具
@mcp.tool()
def say_hello(name: str) -> str:
"""
向指定的人打招呼
"""
return f"你好, {name}!这是来自 MCP 的问候 👋"
if __name__ == "__main__":
# 如果没有这句话,需要通过脚手架启动服务器,也就是
# mcp run server.py (全局安装 mcp[cli]) 或者 uv run mcp run server.py (本地安装 mcp)
mcp.run()
这里你就定义了一个简单的MCP服务了!!!
OpenMCP教程https://openmcp.kirigaya.cn/zh/plugin-tutorial/quick-start/
STEP1:需求确定与快速开发
根据对大模型能力的边界和知识的边界了解估算需要输入的prompt。可以将上面fastmcp的例子输入进去让模型知道mcp怎么用
我现在要开发一个 MCP 服务器让大模型学会调用工具,
请使用 fastmcp 设计一个 excel 文档的 MCP,
使得大模型可以操控 excel 文件,下面是一个 MCP 的开发案例:
from mcp.server.fastmcp import FastMCP
# 创建一个 MCP Server
mcp = FastMCP("HelloMCP")
# 定义一个 prompt
@mcp.prompt()
def hello(name: str) -> str:
"""
hello prompt
"""
return f"hello {name}"
# 定义一个工具
@mcp.tool()
def say_hello(name: str) -> str:
"""
向指定的人打招呼
"""
return f"你好, {name}!这是来自 MCP 的问候 👋"
if __name__ == "__main__":
# 如果没有这句话,需要通过脚手架启动服务器,也就是
# mcp run server.py (全局安装 mcp[cli]) 或者 uv run mcp run server.py (本地安装 mcp)
mcp.run()
我的kimi输出如下:看起来还不错,先直接复制到server_excel.py文件,先运行脚本无报错程序停止不动说明启动成功,后续可以在OpenMCP调试
# server_excel.py
from pathlib import Path
from typing import Optional
from openpyxl import Workbook, load_workbook
from openpyxl.utils import get_column_letter
from mcp.server.fastmcp import FastMCP
# 1. 创建 MCP 服务器
mcp = FastMCP("ExcelMCP")
# ---------- 工具区 ----------
@mcp.tool()
def excel_create(file_path: str, sheet: str = "Sheet1") -> str:
"""
新建一个 Excel 文件(若已存在会覆盖)
:param file_path: 绝对或相对路径,必须以 .xlsx 结尾
:param sheet: 默认工作表名
:return: 成功提示
"""
wb = Workbook()
ws = wb.active
ws.title = sheet
wb.save(file_path)
return f"已创建文件:{Path(file_path).absolute()}"
@mcp.tool()
def excel_read_cell(file_path: str, sheet: str, cell: str) -> str:
"""
读取单个单元格
:param cell: 例如 "A1"
"""
wb = load_workbook(file_path)
ws = wb[sheet]
val = ws[cell].value
return str(val) if val is not None else ""
@mcp.tool()
def excel_write_cell(
file_path: str, sheet: str, cell: str, value: str | int | float
) -> str:
"""
写入单个单元格
"""
wb = load_workbook(file_path)
ws = wb[sheet]
ws[cell] = value
wb.save(file_path)
return f"已写入 {cell} = {value}"
@mcp.tool()
def excel_append_row(
file_path: str, sheet: str, row_data: list[str | int | float]
) -> str:
"""
在最后一行下方追加一行数据
"""
wb = load_workbook(file_path)
ws = wb[sheet]
ws.append(row_data)
wb.save(file_path)
return f"已追加一行:{row_data}"
@mcp.tool()
def excel_sum_column(
file_path: str, sheet: str, col: str, start_row: int = 1
) -> float:
"""
对某列从 start_row 到最后一行求和(只统计数字)
:param col: 列字母,如 "A"
"""
wb = load_workbook(file_path)
ws = wb[sheet]
col_idx = openpyxl.utils.column_index_from_string(col)
total = 0.0
for row in ws.iter_rows(min_row=start_row, max_row=ws.max_row, min_col=col_idx, max_col=col_idx):
cell = row[0]
if isinstance(cell.value, (int, float)):
total += cell.value
return total
# ---------- 可选:Prompt 模板 ----------
@mcp.prompt()
def excel_helper() -> str:
"""
给大模型的一个系统提示,让它知道有哪些 Excel 工具可用
"""
return (
"你是一个 Excel 助手,已连接 ExcelMCP 服务器,拥有以下工具:\n"
"- excel_create: 新建文件\n"
"- excel_read_cell: 读单元格\n"
"- excel_write_cell: 写单元格\n"
"- excel_append_row: 追加行\n"
"- excel_sum_column: 列求和\n"
"当用户提到“打开/新建/读写/追加/求和”等关键词时,直接调用对应工具即可。"
)
# ---------- 启动 ----------
if __name__ == "__main__":
mcp.run()
STEP2:OpenMCP启动!开始调试!
这边简单说明,更多OpenMCP使用请看教程。OpenMCP教程https://openmcp.kirigaya.cn/zh/plugin-tutorial/quick-start/
如果你安装了OpenMCP,直接点这个小圈圈。
一般要调试自建都成功在进行下一步,我报的错误都是文件路径没找到,我看了下每步报错路径都千奇百怪的,感觉没按照这个预设链路来。不知道是不是bug还是模型的问题。我觉得这个问题还好就直接进行下一步了。
STEP3.通过交互测评当前MCP
可以看到还是正常调用MCP生成了文件
STEP4.测试无误后封装prompt
@mcp.prompt
def data2excel_prompt(content: str) -> str:
"""请将非结构化数据整合成excel文件"""
return f"""你是一个擅长擅长分析非结构化数据、并生成excel的的AI,请将输入的资料整理和生成excel文件。{content}"""
STEP5.部署MCP服务器
import { OmAgent } from 'openmcp-sdk/service/sdk';
@Injectable()
export class SlidesService {
/**
* @description 创建 markdown 任务,并返回中途进度
*/
async dataToExcelHandler(id: number, user: User, subscriber: Subscriber<any>) {
// Load configuration, which can be automatically generated after debugging with openmcp client
agent.loadMcpConfig('/path/to/excel-mcp.config.json');
// 从文档数据库中拿当前用户的 content
const content = await this.documentService.getContent(id, user);
const prompt = await agent.getPrompt('data2excel_prompt', { content });
const res = await agent.ainvoke({ messages: prompt });
subscriber.next(toSseData({ done: true, data: res }));
}
}
更多推荐
所有评论(0)