一文幫你 MCP 從 0 到 1 快速入門

MCP 是什麼

MCP 全稱是 Model Context Protocol(模型上下文協議),它是一個專門設計用來讓大語言模型(如 Claude)能夠與外部工具和數據源進行交互的協議。

MCP 能做什麼?

通過 MCP,你可以:

快速入門

在本教程中,我們將構建一個簡單的 MCP 天氣服務器並將其連接到宿主程序 Claude for Desktop。我們將從基礎設置開始,然後逐步過渡到更復雜的用例。

我們要構建什麼

許多 LLM(包括 Claude)目前還沒有獲取天氣預報和嚴重天氣警報的能力。讓我們用 MCP 來解決這個問題!

我們將構建一個提供兩個工具的服務器:get-alerts 和 get-forecast。然後我們將服務器連接到一個 MCP 宿主程序(在本例中是 Claude for Desktop):

服務器可以連接到任何客戶端。我們在這裏選擇 Claude desktop 是爲了簡單起見,我們也有關於構建自己的客戶端的指南。
因爲服務器是本地運行的,MCP 目前只支持桌面宿主程序。遠程宿主程序正在積極開發中。

MCP 核心概念

MCP 服務器可以提供三種主要類型的功能:

本教程主要關注工具,但如果你想了解更多關於資源和提示的內容,我們也有進階教程。

前置知識

本快速入門假設你熟悉:

系統要求

對於 Python,請確保你安裝了 Python 3.9 或更高版本。

配置環境

首先,讓我們安裝 uv 並設置 Python 項目和環境:

curl -LsSf https://astral.sh/uv/install.sh | sh

安裝完成後請重啓終端,以確保 uv 命令可以被正確識別。

現在,讓我們創建並設置項目:

# 爲項目創建新目錄
uv init weather
cd weather

# 創建虛擬環境並激活
uv venv
source .venv/bin/activate

# 安裝依賴
uv add mcp httpx 

# 刪除模板文件
rm hello.py

# 創建我們的文件
mkdir -p src/weather
touch src/weather/__init__.py
touch src/weather/server.py

將以下代碼添加到 pyproject.toml

[build-system]
requires = [ "hatchling",]
build-backend = "hatchling.build"

[project.scripts]
weather = "weather:main"

將以下代碼添加到 __init__.py

from . import server
import asyncio

def main():
    """包的主入口點。"""
    asyncio.run(server.main())

# 可選:在包級別暴露其他重要項
__all__ = ['main''server']

現在讓我們開始構建服務器。

構建服務器

導入包

將以下內容添加到 server.py 的頂部:

from typing import Any
import asyncio
import httpx
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio

設置實例

然後初始化服務器實例和 NWS API 的基礎 URL:

NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

server = Server("weather")

實現工具列表

我們需要告訴客戶端有哪些工具可用。list_tools() 裝飾器會註冊這個處理程序:

@server.list_tools()
asyncdef handle_list_tools() -> list[types.Tool]:
    """
    列出可用的工具。
    每個工具使用 JSON Schema 驗證來指定其參數。
    """
    return [
        types.Tool(
            ,
            description="獲取指定州的天氣預警",
            inputSchema={
                "type""object",
                "properties": {
                    "state": {
                        "type""string",
                        "description""兩字母州代碼(例如 CA、NY)",
                    },
                },
                "required": ["state"],
            },
        ),
        types.Tool(
            ,
            description="獲取指定位置的天氣預報",
            inputSchema={
                "type""object",
                "properties": {
                    "latitude": {
                        "type""number",
                        "description""位置的緯度",
                    },
                    "longitude": {
                        "type""number",
                        "description""位置的經度",
                    },
                },
                "required": ["latitude""longitude"],
            },
        ),
    ]

這裏定義了我們的兩個工具:get-alerts 和 get-forecast

輔助函數

接下來,讓我們添加用於查詢和格式化國家氣象服務 API 數據的輔助函數:

async def make_nws_request(client: httpx.AsyncClient, url: str) -> dict[str, Any] | None:
    """向 NWS API 發送請求並進行適當的錯誤處理。"""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept""application/geo+json"
    }

    try:
        response = await client.get(url, headers=headers, timeout=30.0)
        response.raise_for_status()
        return response.json()
    except Exception:
        returnNone

def format_alert(feature: dict) -> str:
    """將預警特徵格式化爲簡潔的字符串。"""
    props = feature["properties"]
    return (
        f"事件:{props.get('event', '未知')}\n"
        f"區域:{props.get('areaDesc', '未知')}\n"
        f"嚴重程度:{props.get('severity', '未知')}\n"
        f"狀態:{props.get('status', '未知')}\n"
        f"標題:{props.get('headline', '無標題')}\n"
        "---"
    )

實現工具執行

工具執行處理程序負責實際執行每個工具的邏輯。讓我們添加它:

@server.call_tool()
asyncdef handle_call_tool(
    name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    """
    處理工具執行請求。
    工具可以獲取天氣數據並通知客戶端變化。
    """
    ifnot arguments:
        raise ValueError("缺少參數")

    if name == "get-alerts":
        state = arguments.get("state")
        ifnot state:
            raise ValueError("缺少州參數")

        # 將州代碼轉換爲大寫以確保格式一致
        state = state.upper()
        if len(state) != 2:
            raise ValueError("州代碼必須是兩位字母(例如 CA, NY)")

        asyncwith httpx.AsyncClient() as client:
            alerts_url = f"{NWS_API_BASE}/alerts?area={state}"
            alerts_data = await make_nws_request(client, alerts_url)

            ifnot alerts_data:
                return [types.TextContent(type="text", text="獲取預警數據失敗")]

            features = alerts_data.get("features", [])
            ifnot features:
                return [types.TextContent(type="text", text=f"{state} 沒有活躍的預警")]

            # 將每個預警格式化爲簡潔的字符串
            formatted_alerts = [format_alert(feature) for feature in features[:20]] # 僅取前20個預警
            alerts_text = f"{state} 的活躍預警:\n\n" + "\n".join(formatted_alerts)

            return [
                types.TextContent(
                    type="text",
                    text=alerts_text
                )
            ]
    elif name == "get-forecast":
        try:
            latitude = float(arguments.get("latitude"))
            longitude = float(arguments.get("longitude"))
        except (TypeError, ValueError):
            return [types.TextContent(
                type="text",
                text="無效的座標。請提供有效的緯度和經度數字。"
            )]
            
        # 基本座標驗證
        ifnot (-90 <= latitude <= 90) ornot (-180 <= longitude <= 180):
            return [types.TextContent(
                type="text",
                text="無效的座標。緯度必須在 -90 到 90 之間,經度在 -180 到 180 之間。"
            )]

        asyncwith httpx.AsyncClient() as client:
            # 首先獲取網格點
            lat_str = f"{latitude}"
            lon_str = f"{longitude}"
            points_url = f"{NWS_API_BASE}/points/{lat_str},{lon_str}"
            points_data = await make_nws_request(client, points_url)

            ifnot points_data:
                return [types.TextContent(type="text", text=f"獲取座標 {latitude}, {longitude} 的網格點數據失敗。此位置可能不受 NWS API 支持(僅支持美國位置)。")]

            # 從響應中提取預報 URL
            properties = points_data.get("properties", {})
            forecast_url = properties.get("forecast")
            
            ifnot forecast_url:
                return [types.TextContent(type="text", text="從網格點數據獲取預報 URL 失敗")]

            # 獲取預報
            forecast_data = await make_nws_request(client, forecast_url)
            
            ifnot forecast_data:
                return [types.TextContent(type="text", text="獲取預報數據失敗")]

            # 格式化預報週期
            periods = forecast_data.get("properties", {}).get("periods", [])
            ifnot periods:
                return [types.TextContent(type="text", text="沒有可用的預報週期")]

            # 將每個週期格式化爲簡潔的字符串
            formatted_forecast = []
            for period in periods:
                forecast_text = (
                    f"{period.get('name', '未知')}:\n"
                    f"溫度: {period.get('temperature', '未知')}°{period.get('temperatureUnit', 'F')}\n"
                    f"風: {period.get('windSpeed', '未知')} {period.get('windDirection', '')}\n"
                    f"{period.get('shortForecast', '無可用預報')}\n"
                    "---"
                )
                formatted_forecast.append(forecast_text)

            forecast_text = f"{latitude}, {longitude} 的預報:\n\n" + "\n".join(formatted_forecast)

            return [types.TextContent(
                type="text",
                text=forecast_text
            )]
    else:
        raise ValueError(f"未知工具: {name}")

運行服務器

最後,實現主函數來運行服務器:

async def main():
    # 使用標準輸入/輸出流運行服務器
    asyncwith mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_,
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

# 如果你想連接到自定義客戶端,這是必需的
if __name__ == "__main__":
    asyncio.run(main())

你的服務器已經完成!運行 uv run src/weather/server.py 以確認一切正常。

測試你的服務器與 Claude for Desktop

Claude for Desktop 目前不可用於 Linux。Linux 用戶可以繼續進行 [構建客戶端] 教程,以構建一個可以連接到我們剛剛構建的服務器的 MCP 客戶端。

首先,確保你已經安裝了 Claude for Desktop。你可以在這裏 [1] 安裝最新版本。

接下來,在文本編輯器中打開你的 Claude for Desktop App 配置,位於 ~/Library/Application Support/Claude/claude_desktop_config.json

例如,如果你已經安裝了 VS Code:

code ~/Library/Application\ Support/Claude/claude_desktop_config.json

添加此配置(替換父文件夾路徑):

{
    "mcpServers": {
        "weather": {
            "command""uv",
            "args": [
                "--directory",
                "/ABSOLUTE/PATH/TO/PARENT/FOLDER/weather",
                "run",
                "weather"
            ]
        }
    }
}

這告訴 Claude for Desktop:

保存文件,並重新啓動 Claude for Desktop。

測試命令

首先,確保 Claude for Desktop 已經識別到我們在 weather 服務器中暴露的兩個工具。你可以通過查找錘子圖標 <img src="/images/claude-desktop-mcp-hammer-icon.svg" style={{display: 'inline', margin: 0, height: '1.3em'}} /> 來確認:

點擊錘子圖標後,你應該能看到兩個工具:

如果你的服務器沒有被 Claude for Desktop 識別,請查看 [故障排除] 部分獲取調試建議。

現在你可以通過在 Claude for Desktop 中運行以下命令來測試你的服務器:

Sacramento 的天氣怎麼樣?
Texas 有什麼活躍的天氣預警?

背後的原理

當你提出一個問題時:

  1. 客戶端將你的問題發送給 Claude

  2. Claude 分析可用的工具並決定使用哪個工具

  3. 客戶端通過 MCP 服務器執行選定的工具

  4. 結果返回給 Claude

  5. Claude 組織一個自然語言響應

  6. 響應顯示給你!

更多 MCP 資料

參考資料

[1] 

Claud for Desktop: https://claude.ai/download

本文由 Readfog 進行 AMP 轉碼,版權歸原作者所有。
來源https://mp.weixin.qq.com/s/3x-AmVTsIVx9mcrn28bstw