用 Python 開發(fā) MCP 服務(wù)很簡單，完整案例拿走不謝！

發(fā)布于 2025-4-28 00:16

瀏覽

0收藏

模型上下文協(xié)議 MCP 是一種開放協(xié)議，旨在標(biāo)準(zhǔn)化 AI 應(yīng)用程序與外部數(shù)據(jù)源和工具的連接方式。其核心目標(biāo)在于簡化大型語言模型 LLM 與各種上下文和工具的集成，從而解決將多種 LLM 與多種工具相集成的復(fù)雜性問題。

之前的文章介紹了一個挺不錯的 MCP 應(yīng)用案例，

??用 MCP 讓大模型自動批量下載文獻(xiàn)，太香了！??

但你有沒有想過，自己動手打造一個 MCP 服務(wù)呢？想象一下，親手設(shè)計一個智能工具，無論是處理數(shù)據(jù)、調(diào)用外部 API，還是生成自定義報告。這種創(chuàng)造力和掌控力的體驗，會不會更香呢？

MCP 本身是一個通用的協(xié)議，設(shè)計上是為了跨語言和跨平臺的，因此可以使用各種語言來編寫服務(wù)，比如常用的 ??Node.js??。但如果不熟悉 Web 開發(fā)，也想自己寫一寫呢？Python 自然是不二之選。只是 MCP 具體的協(xié)議挺復(fù)雜的，很難快速上手。

不過，現(xiàn)在好了，因為有了 FastMCP，它是一個 Python 軟件開發(fā)工具包 (SDK)，專門設(shè)計用于簡化構(gòu)建 MCP 服務(wù)的過程。它提供了一個高級且符合 Python 語言習(xí)慣的接口，用于定義工具、資源和提示。FastMCP 的核心優(yōu)勢在于其能夠幫助開發(fā)者更輕松地創(chuàng)建符合 MCP 規(guī)范的服務(wù)，而無需深入了解底層的協(xié)議細(xì)節(jié)。

但值得注意的是，你看網(wǎng)上代碼會涉及兩個 FastMCP 包，其中一個是，

https://github.com/jlowin/fastmcp

另一個則是官方的，

https://github.com/modelcontextprotocol/python-sdk

這兩個什么關(guān)系呢？其實是官方收編了上面第一個包，但官方集成的是 fastmcp 的 v1.0 版本。然而，jlowin 繼續(xù)開發(fā) fastmcp，還發(fā)布了 v2.0 版本，其中包含代理和客戶端采樣等新功能。

使用的話這兩個都可以，先看第一個的安裝和導(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ù)。

我們先來看一個最簡單的例子。

./demo/server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Demo ??")

@mcp.tool()
def add_2_numbers(a: int, b: int) -> int:
    """兩個數(shù)字相加"""
    return a + b

if __name__ == "__main__":
    mcp.run(transport='stdio')

這個例子展示了如何使用 FastMCP 庫來創(chuàng)建一個簡單的 MCP 服務(wù)器，它提供一個基本的工具（工具名：??add_2_numbers??），用于兩個數(shù)字相加。這是一個非常簡單的入門示例，適合理解 MCP 服務(wù)器的工作原理和 FastMCP 的基本用法。

首先，從??mcp.server.fastmcp?? 模塊中導(dǎo)入了 FastMCP 類。接著，創(chuàng)建了一個 FastMCP 類的實例，命名為 mcp。FastMCP 是服務(wù)器的核心類，它負(fù)責(zé)管理工具、資源和通信。參數(shù)「Demo ??」是服務(wù)器的名稱，用于標(biāo)識這個 MCP 服務(wù)器。
然后，??@mcp.tool()??? 是一個裝飾器（??decorator???），它告訴 FastMCP 將??add_2_numbers??? 函數(shù)注冊為一個可供客戶端調(diào)用的工具。文檔字符串（"""兩個數(shù)字相加"""）：這是函數(shù)的描述，客戶端（如??MCP Inspector??）會顯示這個描述，幫助用戶理解工具的功能。
這個工具可以被外部客戶端調(diào)用，例如通過 MCP 協(xié)議發(fā)送請求，傳入兩個整數(shù) a 和 b，服務(wù)器會返回它們的和。例如，如果客戶端調(diào)用??add_2_numbers(3, 5)??，服務(wù)器會返回 8。
??mcp.run(transport='stdio')??? 啟動 MCP 服務(wù)器，進(jìn)入監(jiān)聽狀態(tài)，等待客戶端的連接和請求。??transport='stdio'??? 指定了通信方式為標(biāo)準(zhǔn)輸入輸出（??Standard I/O???）。這意味著服務(wù)器通過命令行的標(biāo)準(zhǔn)輸入（stdin）和標(biāo)準(zhǔn)輸出（stdout）與客戶端通信，適合本地開發(fā)和測試。如果需要遠(yuǎn)程通信，可以更改為其他傳輸方式（如??'sse'???，??Server-Sent Events??）。

運(yùn)行效果：

運(yùn)行 ??python ./demo/server.py???，服務(wù)器會啟動并監(jiān)聽 stdio。任何支持 MCP 協(xié)議的客戶端（例如 ??MCP Inspector??? 或 ??Claude Desktop???）都可以連接到它，調(diào)用 ??add_2_numbers?? 工具。

比如，可以通過 ??MCP Inspector?? 測試一下這個工具。

% mcp dev ./demo/server.py
Need to install the following packages:
@modelcontextprotocol/inspector@0.8.2
Ok to proceed? (y) y

用 Python 開發(fā) MCP 服務(wù)很簡單，完整案例拿走不謝！-AI.x社區(qū)

瀏覽器中打開 ??http://127.0.0.1:6274??，可以測試一下上面這個 mcp 服務(wù)里提供的 tool。

用 Python 開發(fā) MCP 服務(wù)很簡單，完整案例拿走不謝！-AI.x社區(qū)

這個 mcp 過于簡單，咱們就不安裝來用了。下面我們手?jǐn)]一個稍微復(fù)雜一點的 mcp 服務(wù)，并安裝上真正使用起來。

記賬 mcp 服務(wù)

我們基于 MCP 開發(fā)了一個輕量級（玩具級）記賬服務(wù) ??Money Track MCP???，旨在幫助用戶管理和跟蹤財務(wù)數(shù)據(jù)。該服務(wù)利用 Python 和 ??FastMCP?? 框架，提供了三種核心功能類型（工具、資源和提示），為用戶提供一個靈活、高效的解決方案來記錄收入和支出、查詢賬戶狀態(tài)以及生成格式化的財務(wù)報告。

項目雖小，但涉及環(huán)境變量的設(shè)置和讀取，用戶數(shù)據(jù)的存儲等多個方面，對于開發(fā)更強(qiáng)大服務(wù)來說是一個不錯的基礎(chǔ)。完整代碼的地址見文末。

核心功能

工具 (??@tool???)：服務(wù)提供了一個名為??record_transaction?? 的工具，允許用戶動態(tài)輸入當(dāng)天的收入和支出，如「今天賺了 500 元，花了 250 元」。工具會自動更新累積的總收入、總支出，并計算當(dāng)前余額。這些數(shù)據(jù)持久化存儲在用戶指定的文件中，確保數(shù)據(jù)不會丟失。
資源 (??@resource???)：通過??get_account_status??? 資源，用戶可以安全地檢索當(dāng)前賬戶的最新狀態(tài)，包括總收入、總支出和余額。資源數(shù)據(jù)存儲在 JSON 文件中，文件路徑可以由用戶通過環(huán)境變量或配置文件自定義，默認(rèn)位于??./accounting_data/accounting_data.json??。
提示 (??@prompt???)：??format_account_report?? 提示負(fù)責(zé)將賬戶狀態(tài)格式化為易讀的報告。

技術(shù)架構(gòu)

框架：基于??FastMCP??，一個高效的 Python 庫，簡化了 MCP 服務(wù)器的開發(fā)。
存儲：數(shù)據(jù)保存在 JSON 文件中，支持用戶自定義存儲路徑（如通過環(huán)境變量??ACCOUNTING_WORKING_DIR?? 或配置文件）。
通信：默認(rèn)使用標(biāo)準(zhǔn)輸入輸出 (??stdio??) 傳輸協(xié)議，方便本地開發(fā)和測試。
日志：支持??INFO??? 和??DEBUG?? 級別的日志記錄，幫助開發(fā)者監(jiān)控服務(wù)運(yùn)行狀態(tài)和調(diào)試問題。

使用場景

??MoneyTrack MCP?? 適用于多種場景，包括但不限于：

個人財務(wù)管理：個人用戶可以通過客戶端記錄日常收支，并隨時查看余額。
企業(yè)/團(tuán)隊記賬：小型企業(yè)或團(tuán)隊可以集成此服務(wù)到更大的財務(wù)系統(tǒng)中，快速統(tǒng)計現(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"]
    }

# 提示：格式化賬戶報告
@mcp.prompt()
def format_account_report(status: Dict) -> str:
    """格式化賬戶狀態(tài)為易讀的報告"""
    returnf"""
    === 賬戶報告 ===
    總收入: ${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ù)啟動 ===")
    logging.info("Money Track MCP 服務(wù)啟動")
    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 配置文件中請參考如下設(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è)置賬本的保存目錄。