從 0 開始實現 MCP-Server

MCP Server 概念

MCP Server 是一箇中間層服務器，它主要負責處理和管理 AI 模型的上下文信息，確保模型能夠高效且準確地理解和響應用戶請求。它作爲應用程序和AI模型之間的橋樑，優化了信息的傳遞和處理過程。

根據 MCP 協議定義，Server 可以提供三種類型的標準能力，Resources、Tools、Prompts，每個 Server 可同時提供者三種類型能力或其中一種。

**Resources：**資源，類似於文件數據讀取，可以是文件資源或是 API 響應返回的內容。比如
**Tools：**工具，第三方服務、功能函數，通過此可控制 LLM 可調用哪些函數。
**Prompts：**提示詞，爲用戶預先定義好的完成特定任務的模板。

MCP 通信方式

MCP（Model Context Protocol）是一種爲了統一大規模模型和工具間通信而設計的協議，它定義了消息格式和通信方式。MCP 協議支持多種傳輸機制，其中包括 stdio、Server-Sent Events（SSE） 和 Streamable HTTP。

Stdio 傳輸（Standard Input/Output）

stdio 傳輸方式是最簡單的通信方式，通常在本地工具之間進行消息傳遞時使用。它利用標準輸入輸出（stdin/stdout）作爲數據傳輸通道，適用於本地進程間的交互。

工作方式：客戶端和服務器通過標準輸入輸出流（stdin/stdout）進行通信。客戶端向服務器發送命令和數據，服務器執行並通過標準輸出返回結果。
應用場景：適用於本地開發、命令行工具、調試環境，或者模型和工具服務在同一進程內運行的情況。

Server-Sent Events（SSE）

SSE 是基於 HTTP 協議的流式傳輸機制，它允許服務器通過 HTTP 單向推送事件到客戶端。SSE 適用於客戶端需要接收服務器推送的場景，通常用於實時數據更新。

工作方式：客戶端通過 HTTP GET 請求建立與服務器的連接，服務器以流式方式持續向客戶端發送數據，客戶端通過解析流數據來獲取實時信息。
應用場景：適用於需要服務器主動推送數據的場景，如實時聊天、天氣預報、新聞更新等。

Streamable HTTP

Streamable HTTP 是 MCP 協議中新引入的一種傳輸方式，它基於 HTTP 協議支持雙向流式傳輸。與傳統的 HTTP 請求響應模型不同，Streamable HTTP 允許服務器在一個長連接中實時向客戶端推送數據，並且可以支持多個請求和響應的流式傳輸。

不過需要注意的是，MCP只提供了Streamable HTTP協議層的支持，也就是規範了MCP客戶端在使用Streamable HTTP通信時的通信規則，而並沒有提供相關的SDK客戶端。開發者在開發Streamable HTTP機制下的客戶端和服務器時，可以使用比如Python httpx庫進行開發。

工作方式：客戶端通過 HTTP``POST 向服務器發送請求，並可以接收流式響應（如 JSON-RPC 響應或 SSE 流）。當請求數據較多或需要多次交互時，服務器可以通過長連接和分批推送的方式進行數據傳輸。
應用場景：適用於需要支持高併發、低延遲通信的分佈式系統，尤其是跨服務或跨網絡的應用。適合高併發的場景，如實時流媒體、在線遊戲、金融交易系統等。

MCP Server 實現流程

在本教程中將帶領大家一起實現一個類似於MCP官網的天氣查詢MCP Server，但是與官網的MCP示例不同的是，官網的天氣查詢僅僅支持美國的州市，無法查詢中國城市的天氣情況。所以，在本教程中，使用的是openweather的免費接口，實現全世界各地的一個通用天氣查詢MCP服務。

業務功能實現

進入OpenWeather官網（https://openweathermap.org/），然後使用自己的信息註冊一個賬號。

接着我們需要申請一個API Keys，用於後期接口校驗。點擊My APIKeys。

默認的情況下，會自動給你生成一個API Keys，你可以直接使用默認生成的API Keys，或者自己重新創建一個API Keys。

雖然兩者都可以，但是建議大家直接使用默認的API Keys即可，因爲創建新的API Keys後，需要等 5 分鐘左右才能生效。默認的API Keys只需要 3 分鐘左右即可使用。

複製自己的API Keys後，點擊菜單欄上的API即可開始選擇自己所需要的服務。

需要注意的是，OpenWeather有很多關於天氣的服務，但是並不是所有服務都是免費的，你需要根據他的描述，選擇自己所需要的服務即可，在這裏我們直接選擇Current Weather Data接口，該接口是免費的。點擊其對應的API doc。

該接口也有很多種請求方式，我們選擇兩種。

通過經緯度，請求對應經緯度的當前天氣情況。
通過城市名稱，查詢對應城市名稱的當前天氣情況。

無論是哪種請求方式，API Key都是必填參數。

我們有了自己的API Key後，可以直接通過瀏覽器請求的方式，驗證當前接口是否可用。直接在瀏覽器中輸入url即可。

比如，我通過指定城市名稱爲 wuhan，查詢武漢對應的天氣情況。

https://api.openweathermap.org/data/2.5/weather?q=wuhan&appid={API key}

注意：{API key} 需要替換爲你自己的。

有以下返回，說明接口是可用的。

如果返回以下內容，說明API Key未生效，或者API Key錯誤。如果是檢查了API Key確定沒有填寫錯誤，那麼請等待幾分鐘後重試。

到這裏，我們已經支持如何通過經緯度和地名獲取天氣了，接下來要做的就是將該服務封裝未MCP Server。

MCP Server 功能編寫

在這裏，我們首先測試stdio通信方式，採用才本地開啓一個MCP Server的方式實現。

首先，使用uv工具，創建項目並安裝相關依賴。這裏我將項目放到D盤的根目錄，在D盤下打開命令提示符。

uv init weather_mcp_server -p 3.10

接着進入uv工程。

cd weather_mcp_server

然後輸入以下命令，創建虛擬環境。

uv venv

激活虛擬環境

.venv\Scripts\activate

由於我電腦中默認會激活一個conda的base虛擬環境，所以再激活uv工程的虛擬環境後，在路徑前面出現了兩個()，故我還需要執行以下命令，退出conda的base虛擬環境。
conda deactivate

安裝依賴

uv add mcp[cli] httpx

依賴準備完成，接下來開始代碼編寫部分內容。

weather.py代碼如下所示：

注：代碼中的{API KEY}部分，請替換爲自己的API Key。

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP

# 初始化FastMCP服務器
mcp = FastMCP("weather")

# 常量
NWS_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36"

# 溫度單位轉換，將開爾文轉化爲攝氏度
def kelvin_to_celsius(kelvin: float) -> float:
    return kelvin - 273.15

asyncdef get_weather_from_cityname(cityname: str) -> dict[str, Any] | None:
    """向openweathermap發送請求並進行適當的錯誤處理。"""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json"
    }
    params = {
        "q": cityname,
        "appid": "{API KEY}"
    }
    asyncwith httpx.AsyncClient() as client:
        try:
            response = await client.get(NWS_API_BASE, headers=headers, params=params)
            response.raise_for_status()
            return response.json()
        except Exception:
            returnNone
        

asyncdef get_weather_from_latitude_longitude(latitude: float, longitude: float) -> dict[str, Any] | None:
    """向openweathermap發送請求並進行適當的錯誤處理。"""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json"
    }
    params = {
        "lat": latitude,
        "lon": longitude,
        "appid": "{API KEY}"
    }
    asyncwith httpx.AsyncClient() as client:
        try:
            response = await client.get(NWS_API_BASE, headers=headers, params=params)
            response.raise_for_status()
            return response.json()
        except Exception:
            returnNone

def format_alert(feature: dict) -> str:
    """將接口返回的天氣信息進行格式化文本輸出"""
    if feature["cod"] == 404:
        return"參數異常，請確認城市名稱是否正確。"
    elif feature["cod"] == 401:
        return"API key 異常，請確認API key是否正確。"
    elif feature["cod"] == 200:
        returnf"""
        City: {feature.get('name', 'Unknown')}
        Weather: {feature.get('weather', [{}])[0].get('description', 'Unknown')}
        Temperature: {kelvin_to_celsius(feature.get('main', {}).get('temp', 0)):.2f}°C
        Humidity: {feature.get('main', {}).get('humidity', 0)}%
        Wind Speed: {feature.get('wind', {}).get('speed', 0):.2f} m/s
        """
    else:
        return"未知錯誤，請稍後再試。"

@mcp.tool()
asyncdef get_weather_from_cityname_tool(city: str) -> str:
    """Get weather information for a city.

    Args:
        city: City name (e.g., "wuhan"). For Chinese cities, please use pinyin
    """
    data = await get_weather_from_cityname(city)
    return format_alert(data)

@mcp.tool()
asyncdef get_weather_from_latitude_longitude_tool(latitude: float, longitude: float) -> str:
    """Get weather information for a location.

    Args:
        latitude: Latitude of the location
        longitude: Longitude of the location
    """
    data = await get_weather_from_latitude_longitude(latitude, longitude)
    return format_alert(data)

if __name__ == "__main__":
    # 初始化並運行服務器
    mcp.run(transport='stdio')

在該代碼中，我們一共定義了兩個Tool：

get_weather_from_cityname_tool：通過城市名稱獲取天氣情況。
get_weather_from_latitude_longitude_tool：通過經緯度獲取天氣情況。

注意，由於MCP協議需要使用到@mcp.tool標記工具函數，所以使用@mcp.tool標記的工具函數，對應的註釋務必寫清楚，後續大模型能夠識別這些工具、工具如何使用以及工具功能，全都是通過這些註釋進行解讀的。所以一個好的MCP Server，其對應的Tool描述也必須要非常的清楚。

MCP Server 測試

我們編輯好代碼後，可以直接在cursor中測試該MCP Server是否可以正常提供功能和被大模型調用。

{
  "mcpServers": {
    "weather": {
          "command": "uv",
          "args": [
              "--directory",
              "D:\\weather_mcp_server",  // 這裏請替換爲自己的項目工程路徑
              "run",
              "weather.py"
          ]
      }
  }
}

設置完成後，可以發現Cursor可以成功加載我們自己寫的weather MCP。

接下來，在對話中測試，看看是否可以調用到天氣查詢服務。

可以看到，我們自己編寫的MCP Server可以成功被Corsor調用。

MCP Server 發佈

前面我們演示的是stdio方式的通信協議，該方式本質上就是在本地允許了一個服務，然後通過MCP Client去調用本地的服務實現的。

這種方式的缺點是，無法將自己的MCP Server，在不把源代碼給別人的情況下，共享給其他人使用。這種方式在企業中肯定是不能夠被允許的，源代碼是企業的命脈。此時，SSE通信方式就派上用場了，可以使用SSE的通信方式將自己的MCP Server部署在服務器，然後其他所有的MCP Client要調用MCP Server對應的服務時，就無需在本地去執行MCP Server服務了。

如果需要其他所有人都可以訪問當你的MCP Server，就需要將自己的MCP Server配置到公網服務器。

在公網中配置環境的方式與前面的流程一致，需要按照以下方式修改代碼即可。

將weather_sse.py代碼修改爲以下內容，配置爲sse通信協議。

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP

# 初始化FastMCP服務器
# mcp = FastMCP("weather")
mcp = FastMCP(
    ,
    host="0.0.0.0",
    port=8000,
    description="通過城市名稱（拼音）或經緯度獲取天氣信息",
    sse_path="/sse"
)

# 常量
NWS_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36"

# 溫度單位轉換，將開爾文轉化爲攝氏度
def kelvin_to_celsius(kelvin: float) -> float:
    return kelvin - 273.15

asyncdef get_weather_from_cityname(cityname: str) -> dict[str, Any] | None:
    """向openweathermap發送請求並進行適當的錯誤處理。"""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json"
    }
    params = {
        "q": cityname,
        "appid": "24ecadbe4bb3d55cb1f06ea48a41ac51"
    }
    asyncwith httpx.AsyncClient() as client:
        try:
            response = await client.get(NWS_API_BASE, headers=headers, params=params)
            response.raise_for_status()
            return response.json()
        except Exception:
            returnNone
        

asyncdef get_weather_from_latitude_longitude(latitude: float, longitude: float) -> dict[str, Any] | None:
    """向openweathermap發送請求並進行適當的錯誤處理。"""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json"
    }
    params = {
        "lat": latitude,
        "lon": longitude,
        "appid": "24ecadbe4bb3d55cb1f06ea48a41ac51"
    }
    asyncwith httpx.AsyncClient() as client:
        try:
            response = await client.get(NWS_API_BASE, headers=headers, params=params)
            response.raise_for_status()
            return response.json()
        except Exception:
            returnNone

def format_alert(feature: dict) -> str:
    """將接口返回的天氣信息進行格式化文本輸出"""
    if feature["cod"] == 404:
        return"參數異常，請確認城市名稱是否正確。"
    elif feature["cod"] == 401:
        return"API key 異常，請確認API key是否正確。"
    elif feature["cod"] == 200:
        returnf"""
        City: {feature.get('name', 'Unknown')}
        Weather: {feature.get('weather', [{}])[0].get('description', 'Unknown')}
        Temperature: {kelvin_to_celsius(feature.get('main', {}).get('temp', 0)):.2f}°C
        Humidity: {feature.get('main', {}).get('humidity', 0)}%
        Wind Speed: {feature.get('wind', {}).get('speed', 0):.2f} m/s
        """
    else:
        return"未知錯誤，請稍後再試。"

@mcp.tool()
asyncdef get_weather_from_cityname_tool(city: str) -> str:
    """Get weather information for a city.

    Args:
        city: City name (e.g., "wuhan"). For Chinese cities, please use pinyin
    """
    data = await get_weather_from_cityname(city)
    return format_alert(data)

@mcp.tool()
asyncdef get_weather_from_latitude_longitude_tool(latitude: float, longitude: float) -> str:
    """Get weather information for a location.

    Args:
        latitude: Latitude of the location
        longitude: Longitude of the location
    """
    data = await get_weather_from_latitude_longitude(latitude, longitude)
    return format_alert(data)

if __name__ == "__main__":
    # 初始化並運行服務器
    # mcp.run(transport='stdio')
    print("Starting server...")
    mcp.run(transport='sse')

在公網中執行該腳本，開啓MCP SSE Server。執行完成後，可以看到如下圖所示的打印結果。

此時，再次使用Cursor測試SSE MCP Server服務是否可以正常調用。

在Cursor的MCP配置中，修改配置文件如下所示：

{
  "mcpServers": {
    "weather": {
      "url": "http://{你的公網IP}:8000/sse"
    }
  }
}

配置完成後，最好重啓Cursor一次，因爲前面我們加載過相同名字的MCP服務。

重啓完成後，可以看到Cursor也是可以正常識別到我們的MCP Server。

測試功能是否可以正常調用。

功能也可以正常調用。

服務器也有請求響應。

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/d9rJfxmm0W5QYu5s8KWGzw