用 Python 開發(fā) MCP 服務(wù)很簡(jiǎn)單,完整案例拿走不謝!
模型上下文協(xié)議 MCP 是一種開放協(xié)議,旨在標(biāo)準(zhǔn)化 AI 應(yīng)用程序與外部數(shù)據(jù)源和工具的連接方式。其核心目標(biāo)在于簡(jiǎn)化大型語(yǔ)言模型 LLM 與各種上下文和工具的集成,從而解決將多種 LLM 與多種工具相集成的復(fù)雜性問題。
之前的文章介紹了一個(gè)挺不錯(cuò)的 MCP 應(yīng)用案例,
??用 MCP 讓大模型自動(dòng)批量下載文獻(xiàn),太香了!??
但你有沒有想過,自己動(dòng)手打造一個(gè) MCP 服務(wù)呢?想象一下,親手設(shè)計(jì)一個(gè)智能工具,無論是處理數(shù)據(jù)、調(diào)用外部 API,還是生成自定義報(bào)告。這種創(chuàng)造力和掌控力的體驗(yàn),會(huì)不會(huì)更香呢?
MCP 本身是一個(gè)通用的協(xié)議,設(shè)計(jì)上是為了跨語(yǔ)言和跨平臺(tái)的,因此可以使用各種語(yǔ)言來編寫服務(wù),比如常用的 ??Node.js?
?。但如果不熟悉 Web 開發(fā),也想自己寫一寫呢?Python 自然是不二之選。只是 MCP 具體的協(xié)議挺復(fù)雜的,很難快速上手。
不過,現(xiàn)在好了,因?yàn)橛辛?FastMCP,它是一個(gè) Python 軟件開發(fā)工具包 (SDK),專門設(shè)計(jì)用于簡(jiǎn)化構(gòu)建 MCP 服務(wù)的過程。它提供了一個(gè)高級(jí)且符合 Python 語(yǔ)言習(xí)慣的接口,用于定義工具、資源和提示。FastMCP 的核心優(yōu)勢(shì)在于其能夠幫助開發(fā)者更輕松地創(chuàng)建符合 MCP 規(guī)范的服務(wù),而無需深入了解底層的協(xié)議細(xì)節(jié)。
但值得注意的是,你看網(wǎng)上代碼會(huì)涉及兩個(gè) FastMCP 包,其中一個(gè)是,
https://github.com/jlowin/fastmcp
另一個(gè)則是官方的,
https://github.com/modelcontextprotocol/python-sdk
這兩個(gè)什么關(guān)系呢?其實(shí)是官方收編了上面第一個(gè)包,但官方集成的是 fastmcp 的 v1.0 版本。然而,jlowin 繼續(xù)開發(fā) fastmcp,還發(fā)布了 v2.0 版本,其中包含代理和客戶端采樣等新功能。
使用的話這兩個(gè)都可以,先看第一個(gè)的安裝和導(dǎo)入,
pip install fastmcp
from fastmcp import FastMCP
官方的安裝和導(dǎo)入如下所示,
pip install mcp
from mcp.server.fastmcp import FastMCP
本文使用官方 Python SDK 里的 FastMCP 來構(gòu)建自己的 MCP 服務(wù)。
我們先來看一個(gè)最簡(jiǎn)單的例子。
./demo/server.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Demo ??")
@mcp.tool()
def add_2_numbers(a: int, b: int) -> int:
"""兩個(gè)數(shù)字相加"""
return a + b
if __name__ == "__main__":
mcp.run(transport='stdio')
這個(gè)例子展示了如何使用 FastMCP 庫(kù)來創(chuàng)建一個(gè)簡(jiǎn)單的 MCP 服務(wù)器,它提供一個(gè)基本的工具(工具名:??add_2_numbers?
?),用于兩個(gè)數(shù)字相加。這是一個(gè)非常簡(jiǎn)單的入門示例,適合理解 MCP 服務(wù)器的工作原理和 FastMCP 的基本用法。
- 首先,從?
?mcp.server.fastmcp?
? 模塊中導(dǎo)入了 FastMCP 類。接著,創(chuàng)建了一個(gè) FastMCP 類的實(shí)例,命名為 mcp。FastMCP 是服務(wù)器的核心類,它負(fù)責(zé)管理工具、資源和通信。參數(shù)「Demo ??」是服務(wù)器的名稱,用于標(biāo)識(shí)這個(gè) MCP 服務(wù)器。 - 然后,?
?@mcp.tool()?
?? 是一個(gè)裝飾器(??decorator?
??),它告訴 FastMCP 將??add_2_numbers?
?? 函數(shù)注冊(cè)為一個(gè)可供客戶端調(diào)用的工具。文檔字符串("""兩個(gè)數(shù)字相加"""):這是函數(shù)的描述,客戶端(如??MCP Inspector?
?)會(huì)顯示這個(gè)描述,幫助用戶理解工具的功能。 - 這個(gè)工具可以被外部客戶端調(diào)用,例如通過 MCP 協(xié)議發(fā)送請(qǐng)求,傳入兩個(gè)整數(shù) a 和 b,服務(wù)器會(huì)返回它們的和。例如,如果客戶端調(diào)用?
?add_2_numbers(3, 5)?
?,服務(wù)器會(huì)返回 8。 - ?
?mcp.run(transport='stdio')?
?? 啟動(dòng) MCP 服務(wù)器,進(jìn)入監(jiān)聽狀態(tài),等待客戶端的連接和請(qǐng)求。??transport='stdio'?
?? 指定了通信方式為標(biāo)準(zhǔn)輸入輸出(??Standard I/O?
??)。這意味著服務(wù)器通過命令行的標(biāo)準(zhǔn)輸入(stdin)和標(biāo)準(zhǔn)輸出(stdout)與客戶端通信,適合本地開發(fā)和測(cè)試。如果需要遠(yuǎn)程通信,可以更改為其他傳輸方式(如??'sse'?
??,??Server-Sent Events?
?)。
運(yùn)行效果:
運(yùn)行 ??python ./demo/server.py?
??,服務(wù)器會(huì)啟動(dòng)并監(jiān)聽 stdio。任何支持 MCP 協(xié)議的客戶端(例如 ??MCP Inspector?
?? 或 ??Claude Desktop?
??)都可以連接到它,調(diào)用 ??add_2_numbers?
? 工具。
比如,可以通過 ??MCP Inspector?
? 測(cè)試一下這個(gè)工具。
% mcp dev ./demo/server.py
Need to install the following packages:
@modelcontextprotocol/inspector@0.8.2
Ok to proceed? (y) y
瀏覽器中打開 ??http://127.0.0.1:6274?
?,可以測(cè)試一下上面這個(gè) mcp 服務(wù)里提供的 tool。
這個(gè) mcp 過于簡(jiǎn)單,咱們就不安裝來用了。下面我們手?jǐn)]一個(gè)稍微復(fù)雜一點(diǎn)的 mcp 服務(wù),并安裝上真正使用起來。
記賬 mcp 服務(wù)
我們基于 MCP 開發(fā)了一個(gè)輕量級(jí)(玩具級(jí))記賬服務(wù) ??Money Track MCP?
??,旨在幫助用戶管理和跟蹤財(cái)務(wù)數(shù)據(jù)。該服務(wù)利用 Python 和 ??FastMCP?
? 框架,提供了三種核心功能類型(工具、資源和提示),為用戶提供一個(gè)靈活、高效的解決方案來記錄收入和支出、查詢賬戶狀態(tài)以及生成格式化的財(cái)務(wù)報(bào)告。
項(xiàng)目雖小,但涉及環(huán)境變量的設(shè)置和讀取,用戶數(shù)據(jù)的存儲(chǔ)等多個(gè)方面,對(duì)于開發(fā)更強(qiáng)大服務(wù)來說是一個(gè)不錯(cuò)的基礎(chǔ)。完整代碼的地址見文末。
核心功能
- 工具 (?
?@tool?
??):服務(wù)提供了一個(gè)名為??record_transaction?
? 的工具,允許用戶動(dòng)態(tài)輸入當(dāng)天的收入和支出,如「今天賺了 500 元,花了 250 元」。工具會(huì)自動(dòng)更新累積的總收入、總支出,并計(jì)算當(dāng)前余額。這些數(shù)據(jù)持久化存儲(chǔ)在用戶指定的文件中,確保數(shù)據(jù)不會(huì)丟失。 - 資源 (?
?@resource?
??):通過??get_account_status?
?? 資源,用戶可以安全地檢索當(dāng)前賬戶的最新狀態(tài),包括總收入、總支出和余額。資源數(shù)據(jù)存儲(chǔ)在 JSON 文件中,文件路徑可以由用戶通過環(huán)境變量或配置文件自定義,默認(rèn)位于??./accounting_data/accounting_data.json?
?。 - 提示 (?
?@prompt?
??):??format_account_report?
? 提示負(fù)責(zé)將賬戶狀態(tài)格式化為易讀的報(bào)告。
技術(shù)架構(gòu)
- 框架:基于?
?FastMCP?
?,一個(gè)高效的 Python 庫(kù),簡(jiǎn)化了 MCP 服務(wù)器的開發(fā)。 - 存儲(chǔ):數(shù)據(jù)保存在 JSON 文件中,支持用戶自定義存儲(chǔ)路徑(如通過環(huán)境變量?
?ACCOUNTING_WORKING_DIR?
? 或配置文件)。 - 通信:默認(rèn)使用標(biāo)準(zhǔn)輸入輸出 (?
?stdio?
?) 傳輸協(xié)議,方便本地開發(fā)和測(cè)試。 - 日志:支持?
?INFO?
?? 和??DEBUG?
? 級(jí)別的日志記錄,幫助開發(fā)者監(jiān)控服務(wù)運(yùn)行狀態(tài)和調(diào)試問題。
使用場(chǎng)景
??MoneyTrack MCP?
? 適用于多種場(chǎng)景,包括但不限于:
- 個(gè)人財(cái)務(wù)管理:個(gè)人用戶可以通過客戶端記錄日常收支,并隨時(shí)查看余額。
- 企業(yè)/團(tuán)隊(duì)記賬:小型企業(yè)或團(tuán)隊(duì)可以集成此服務(wù)到更大的財(cái)務(wù)系統(tǒng)中,快速統(tǒng)計(jì)現(xiàn)金流。
- 教育與開發(fā)示例:開發(fā)者可以作為 MCP 服務(wù)的學(xué)習(xí)示例,了解工具、資源和提示的協(xié)同工作。
具體代碼如下,
# ./src/server.py
from mcp.server.fastmcp import FastMCP
import os
import json
from typing import Dict, Optional
import logging
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 從環(huán)境變量或默認(rèn)值獲取工作目錄
DEFAULT_WORKING_DIR = "~/accounting_data"
WORKING_DIR = os.getenv("ACCOUNTING_WORKING_DIR", DEFAULT_WORKING_DIR)
# 確保目錄存在
os.makedirs(WORKING_DIR, exist_ok=True)
# 數(shù)據(jù)文件路徑
DATA_FILE = os.path.join(WORKING_DIR, "accounting_data.json")
# 初始數(shù)據(jù)(如果文件不存在)
INITIAL_DATA = {
"total_income": 0,
"total_expense": 0,
"balance": 0
}
def load_data() -> Dict:
"""從文件中加載數(shù)據(jù),如果文件不存在則創(chuàng)建默認(rèn)數(shù)據(jù)"""
try:
if os.path.exists(DATA_FILE):
with open(DATA_FILE, "r", encoding="utf-8") as f:
return json.load(f)
else:
with open(DATA_FILE, "w", encoding="utf-8") as f:
json.dump(INITIAL_DATA, f, indent=4)
return INITIAL_DATA
except Exception as e:
logger.error(f"Failed to load data: {str(e)}")
return INITIAL_DATA
def save_data(data: Dict) -> None:
"""將數(shù)據(jù)保存到文件"""
try:
with open(DATA_FILE, "w", encoding="utf-8") as f:
json.dump(data, f, indent=4)
except Exception as e:
logger.error(f"Failed to save data: {str(e)}")
# 創(chuàng)建 MCP 服務(wù)器
mcp = FastMCP("money-track-mcp", log_level="ERROR")
# 工具:記錄收入和支出
@mcp.tool()
def record_transaction(income: Optional[int] = 0, expense: Optional[int] = 0) -> Dict:
"""記錄今天的收入和支出,更新賬戶余額"""
ifnot isinstance(income, (int, float)) ornot isinstance(expense, (int, float)):
return {"error": "Income and expense must be numbers"}
data = load_data()
data["total_income"] += income
data["total_expense"] += expense
data["balance"] = data["total_income"] - data["total_expense"]
save_data(data)
return {
"message": "Transaction recorded successfully",
"total_income": data["total_income"],
"total_expense": data["total_expense"],
"balance": data["balance"]
}
# 資源:獲取當(dāng)前賬戶狀態(tài)
@mcp.resource("accounting://status")
def get_account_status() -> Dict:
"""獲取當(dāng)前賬戶的收入、支出和余額"""
data = load_data()
return {
"total_income": data["total_income"],
"total_expense": data["total_expense"],
"balance": data["balance"]
}
# 提示:格式化賬戶報(bào)告
@mcp.prompt()
def format_account_report(status: Dict) -> str:
"""格式化賬戶狀態(tài)為易讀的報(bào)告"""
returnf"""
=== 賬戶報(bào)告 ===
總收入: ${status["total_income"]:.2f}
總支出: ${status["total_expense"]:.2f}
當(dāng)前余額: ${status["balance"]:.2f}
================
"""
def run_server():
"""運(yùn)行 MCP 服務(wù)器"""
print("=== Money Track MCP 服務(wù)啟動(dòng) ===")
logging.info("Money Track MCP 服務(wù)啟動(dòng)")
print(f"當(dāng)前工作目錄: {os.getcwd()}")
mcp.run(transport='stdio')
安裝和運(yùn)行
git clone https://github.com/mathinml/mcp_money_track.git
cd mcp_money_track
pip install mcp
pip install .
打開你喜歡的 mcp 客戶端,比如 ??Cline?
??,??Roo Code?
??, ??Cursor?
? 等等。mcp 配置文件中請(qǐng)參考如下設(shè)置,
"money-track-mcp": {
"command": "/opt/anaconda3/bin/python",
"args": [
"-m",
"mcp_money_track"
],
"env": {
"ACCOUNTING_WORKING_DIR": "/Users/xyz/account"
}
},
環(huán)境變量 ??"ACCOUNTING_WORKING_DIR"?
? 是用于設(shè)置賬本的保存目錄。
1、Cline
點(diǎn)擊 Approve 以后,
再次輸入:我今天收入才 500 元,花掉了 1500 元。
沒多說啥,它自動(dòng)領(lǐng)會(huì)要記賬啦,請(qǐng)看結(jié)果。
打開保存目錄里的賬本看一看,
2、Roo Code
然后在另一個(gè)客戶端里試試,VS Code + 插件 Roo Code。
換套說辭,咱不提記賬不記賬,只跟它說:我想買個(gè)電腦,1 萬(wàn)元左右,看看我的余額還夠嗎?
點(diǎn)擊 Approve,如果覺得每次要點(diǎn)太煩人,也可以在 Auto-approve 里設(shè)定總是允許。
余額不夠啦,誰(shuí)打賞一點(diǎn)讓我湊個(gè)整唄。
3、Cherry Studio
在 Cherry Studio 里設(shè)置也是類似的,如圖。
然后,在聊天界面一定要在 MCP 服務(wù)器里選中這個(gè)服務(wù)。
大模型這里我們選用 OpenRoute 提供的免費(fèi) Gemini 2.5 Pro,
也還行,速度上感覺比前面兩個(gè)慢一丟丟。
好了,通過這個(gè)例子,應(yīng)該已經(jīng)明白如何使用 Python 來寫一個(gè)簡(jiǎn)單但完整的 mcp 服務(wù)。
代碼:??https://github.com/mathinml/mcp_money_track?
?
本文轉(zhuǎn)載自??機(jī)器學(xué)習(xí)與數(shù)學(xué)??,作者:大師兄
