譯者 | 朱先忠
審校 | 重樓
簡介
什么是模型上下文協議（Model Context Protocol）？讓我們深入了解MCP背后的概念。以下是官方MCP文檔對MCP的介紹：
“MCP是一種開放協議，它標準化了應用程序向LLM提供上下文的方式。可以將MCP視為AI應用程序的USB-C端口。正如USB-C提供了一種將你的設備連接到各種外圍設備和配件的標準化方式一樣，MCP提供了一種將AI模型連接到不同數據源和工具的標準化方式。”
讓我來解釋一下。假設你正在構建與不同語言模型和AI系統配合使用的AI代理，其中每個模型對工具的理解方式都不同。你已經編寫了代碼來使你的AI代理能夠針對一個特定的AI模型進行構建，假設你的系統構架如下圖所示：
假設你將來可能希望切換到具有不同架構和工具定義方法的另一個AI模型，那么你必須回去重新編寫工具以適應這種新的AI模型架構和方法，假設新架構如下圖所示：
我想，作為程序員，你已經看到了這里的問題——這是不可擴展的。如果我們可以編寫一次工具，然后能夠將其與任何AI模型架構連接起來，而不必擔心這個AI模型架構在后臺如何工作，那會怎樣呢？
這會為我們省去很多麻煩，不是嗎？是的，它不僅可擴展，我們還可以連接任何我們想要的AI模型！
為什么選擇MCP？
你可能會想，我們剛剛引入了另一層（MCP層），另一層就意味著更復雜嗎？是的，但增加這一層的好處遠遠大于壞處。下面是官方文檔的說明：
“MCP可幫助你在LLM之上構建代理和復雜的工作流。LLM通常需要與數據和工具集成，而MCP可提供：
- 你的LLM可以直接插入不斷增加的預構建集成列表
- 在LLM提供商和供應商之間切換的靈活性
- 保護基礎架構內數據的最佳實踐”
MCP的總體架構
總體來說，MCP架構遵循客戶端-服務器架構。我們可以讓一個客戶端連接到多個服務器（MCP服務器）。
現在，讓我們來分析一下上面的圖形架構：
- MCP主機：頂部的“主機（代理、工具）”框代表想要通過模型上下文協議訪問數據的程序。
- MCP客戶端：通過MCP協議直接與MCP服務器（A、B、C）連接的客戶端。
- MCP服務器：用三個框表示（MCP服務器A、B、C），每個框連接到不同的服務。
- 本地數據源：文件系統和本地Postgres數據庫。
- 遠程服務：虛擬私有云之外的Postgres存儲。該架構顯示了一個包含MCP基礎設施的VPC（虛擬私有云）。其中，主機與多個MCP服務器通信，每個服務器處理特定的服務集成。
MCP中的核心概念
- 資源：客戶端可以訪問和讀取的數據對象（類似于文件或API響應）。
- 工具：LLM可以觸發的可執行函數（需要用戶權限）。
- 提示：現成的文本模板，旨在幫助用戶完成特定任務。
創建你的第一個MCP服務器
對于我們的第一個MCP服務器，我想直接在官方文檔上創建一個天氣MCP服務器，這只是為了讓我們更迅速地了解MCP的方式。然后，我們將此服務器連接到一個LangChain代理。
如果愿意，你可以按照官方文檔中的說明進行操作，我也會在本文中提供相關步驟。
安裝環境
在本文案例中，我們選擇使用uv包管理器，它是推薦的包管理器，而且速度非常快，所以我會堅持使用它。通過運行下面的命令來安裝它：
curl -LsSf https://astral.sh/uv/install.sh | sh
至此，我已經成功地將它安裝在我的機器上。如果這是你第一次安裝它，你可能需要重新啟動你的終端。
簡言之，我正在使用Ma/Linux命令。如果你使用的是Windows，你可以按照官方文檔中的Powershell命令進行操作。
# 為我們的項目創建一個新目錄
uv init weather
cd weather
# 創建虛擬環境并激活它
uv venv
source .venv/bin/activate
# 安裝依賴項
uv add "mcp[cli]" httpx
# 創建我們的服務器文件
touch weather.py
完成后，你可以在你最喜歡的IDE中打開目錄。我將使用VSCode；如果你愿意，也可以使用Cursor或者任何其他IDE。
code .編寫服務器端代碼
對于本文中的代碼，我將使用官方文檔中的代碼。在此，非常感謝MCP團隊提供的代碼。
首先，我們將實例化FastMCP類，這有助于大多數工具創建邏輯，例如來自工具函數的文檔字符串的工具描述以及函數類型提示。
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
#初始化FastMCP服務器
mcp = FastMCP("weather")
# 指定常量
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"輔助函數
我們還將創建幾個輔助函數，用于幫助格式化來自API的數據。
async def make_nws_request(url: str) -> dict[str, Any] | None:
"""使用恰當的錯誤處理方式向NWS API提出請求。"""
headers = {
"User-Agent": USER_AGENT,
"Accept": "application/geo+json"
}
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, headers=headers, timeout=30.0)
response.raise_for_status()
return response.json()
except Exception:
return None
def format_alert(feature: dict) -> str:
"""將報警功能格式化為可讀的字符串。"""
props = feature["properties"]
return f"""
Event: {props.get('event', 'Unknown')}
Area: {props.get('areaDesc', 'Unknown')}
Severity: {props.get('severity', 'Unknown')}
Description: {props.get('description', 'No description available')}
Instructions: {props.get('instruction', 'No specific instructions provided')}
"""創建工具
現在，我們將使用Python中的裝飾器在MCP服務器下創建實際的工具mcp.tool()。
@mcp.tool()
async def get_alerts(state: str) -> str:
"""獲取美國一個州的天氣警報。
參數:
state: 兩個字母的美國州代碼（例如CA，NY)
"""
url = f"{NWS_API_BASE}/alerts/active/area/{state}"
data = await make_nws_request(url)
if not data or "features" not in data:
return "Unable to fetch alerts or no alerts found."
if not data["features"]:
return "No active alerts for this state."
alerts = [format_alert(feature) for feature in data["features"]]
return "\n---\n".join(alerts)
@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
"""獲取一個地點的天氣預報。
參數:
latitude: 位置的緯度
longitude: 位置的經度
"""
# 首先獲取預測網格端點
points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
points_data = await make_nws_request(points_url)
if not points_data:
return "Unable to fetch forecast data for this location."
# 從端點響應中獲取預測的URL
forecast_url = points_data["properties"]["forecast"]
forecast_data = await make_nws_request(forecast_url)
if not forecast_data:
return "Unable to fetch detailed forecast."
# 將時間范圍格式化為可讀的預測
periods = forecast_data["properties"]["periods"]
forecasts = []
for period in periods[:5]: # 只顯示未來5個時段
forecast = f"""
{period['name']}:
Temperature: {period['temperature']}°{period['temperatureUnit']}
Wind: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}
"""
forecasts.append(forecast)
return "\n---\n".join(forecasts)啟動服務器
一旦我們完成上面所有這些工作，我們就可以在腳本中添加入口點來執行MCP服務器。現在，在腳本文件weather.py的底部添加以下代碼：
if __name__ == "__main__" :
# 初始化并運行服務器
mcp.run(transport= 'stdio' )從上面的代碼中我們指定了stdio，這是什么意思？
HTTP中的STDIO（標準輸入/輸出）是指使用HTTP連接時輸入和輸出數據的標準流。在Web服務器和HTTP環境中：
- 標準輸入（stdin）：用于接收發送到服務器的數據，如POST請求數據。
- 標準輸出（stdout）：用于將響應數據發送回客戶端。
- 標準錯誤（stderr）：用于記錄錯誤和調試信息在構建使用命令行界面的HTTP服務器或服務時，STDIO提供了一種通過標準Unix風格流傳輸HTTP請求/響應數據的方法，允許與其他命令行工具和進程集成。
我們還可以指定SSE通信方式： - HTTP技術允許服務器將更新推送到客戶端。
- 單向通信（僅限服務器到客戶端）。
- 保持連接暢通以獲取實時更新。
- 比WebSocket更簡單。
- 用于通知、數據饋送和流更新。
完成后，導航到weather.py腳本所在的位置并在終端中運行以下命令：
uv run weather.py
除了看不到輸出內容之外，這表明服務器正在運行，或者你可以更新腳本以顯示某些內容（如果你愿意）。
連接到客戶端
你可以使用不同的客戶端連接到此服務器，例如Claude桌面客戶端、Cursor和許多其他客戶端。你可以在此處閱讀更多相關信息。
LangChain代理MCP客戶端
我想創建一個自定義LangChain代理來連接我們正在運行的MCP服務器。為此，我們必須安裝langchain-mcp-adapters。你可以運行以下命令。
首先，停止天氣腳本并運行以下命令：
uv add ipykernel原因是我將在VScode中使用一個筆記本文件作為LangChain代理。
安裝完成后，繼續再次運行天氣MCP服務器腳本：
uv run weather.py我還在與我們的文件weather.py相同的目錄中繼續創建另一個文件client.ipynb。
然后，你可以運行下面的命令來安裝LangChain MCP適配器：
!uv add langchain-mcp-adapters安裝完成后，我們可以安裝langchain-anthropic和LangGraph客戶端。
!uv add langgraph langchain-anthropic python-dotenv
加載環境變量
首先，我們需要一個Anthropic API密鑰。
一旦你獲得Anthropic API密鑰，你就可以添加.env文件，該文件應該位于你的項目的根目錄中。
ANTHROPIC_API_KEY =sk-xxxxxxxx請確保用實際的API密鑰替換上面的占位符。
接下來，我們可以使用以下方式加載API密鑰：
from dotenv import load_dotenv
load_dotenv()
import os
api_key=os.environ.get("ANTHROPIC_API_KEY")
為stdio連接創建服務器參數
現在，我們可以創建與我們正在運行的MCP服務器的stdio連接服務器參數。
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic讓我們繼續創建模型，我將使用前面提到的Anthropic。我們再次提及了如何使用多個LLM提供商的問題。
model = ChatAnthropic(model="claude-3-5-sonnet-20241022", api_key=api_key)你可以在下面的鏈接處找到有關Anthropic聊天模型的更多信息：
All models overview - Anthropic
server_params = StdioServerParameters(
command= "python" ,
# 確保更新為math_server.py文件的完整絕對路徑
args=[ "./weather.py" ],
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# 初始化連接
await session.initialize()
# 獲取工具
tools = await load_mcp_tools(session)
# 創建并運行代理
agent = create_react_agent(model, tools)
agent_response = await agent.ainvoke({ "messages" : "加州目前的天氣怎么樣" })然后，運行下面命令：
agent_response
漂亮的輸出
現在，讓我們讓輸出看起來更美觀一點：
from IPython.display import display, Markdown
from langchain_core.messages import HumanMessage, ToolMessage, AIMessage
for response in agent_response["messages"]:
user = ""
if isinstance(response, HumanMessage):
user = "**User**"
elif isinstance(response, ToolMessage):
user = "**Tool**"
elif isinstance(response, AIMessage):
user = "**AI**"
if isinstance(response.content, list):
display(Markdown(f'{user}: {response.content[0].get("text", "")}'))
continue
display(Markdown(f"{user}: {response.content}"))
在SSE協議上運行MCP（響應流）
我希望能夠流式傳輸響應，為此我們需要將trasportMCP的類型設置為sse。
為此，停止服務器（MCP服務器）并更改這部分代碼：
if __name__ == "__main__" :
# 初始化并運行服務器
mcp.run(transport= 'sse' )一旦完成，請使用以下命令再次運行代碼：
uv run weather.py
返回筆記本文件中，添加以下語句：
from langchain_mcp_adapters.client import MultiServerMCPClient要測試它，你可以使用：
async with MultiServerMCPClient(
{
"weather": {
"url": "http://localhost:8000/sse",
"transport": "sse",
}
}
) as client:
agent = create_react_agent(model, client.get_tools())
agent_response = await agent.ainvoke({"messages": "what is the weather in nyc?"})
for response in agent_response["messages"]:
user = ""
if isinstance(response, HumanMessage):
user = "**User**"
elif isinstance(response, ToolMessage):
user = "**Tool**"
elif isinstance(response, AIMessage):
user = "**AI**"
if isinstance(response.content, list):
display(Markdown(f'{user}: {response.content[0].get("text", "")}'))
continue
display(Markdown(f"{user}: {response.content}"))
流式響應
我希望能夠實時流式傳輸響應。為此，讓我們編寫以下代碼行：
async with MultiServerMCPClient(
{
"weather" : {
"url" : "http://localhost:8000/sse" ,
"transport" : "sse" ,
}
}
) as client:
agent = create_react_agent(model, client.get_tools())
# 流式傳輸響應塊
async for chunk in agent.astream({ "messages" : "what is the weather in nyc!" }):
# 從AddableUpdatesDict結構中提取消息內容
if 'agent' in chunk and 'messages' in chunk[ 'agent' ]:
for message in chunk[ 'agent' ][ 'messages' ]:
if isinstance (message, AIMessage):
# 處理不同的內容格式
if isinstance (message.content, list ):
# 對于帶有文本和工具使用的結構化內容
for item in message.content:
if isinstance (item, dict ) and 'text' in item:
display(Markdown( f"**AI**: {item[ 'text' ]} " ))
else :
# 對于簡單文本內容
display(Markdown( f"**AI**: {message.content} " ))
elif 'tools' in chunk and 'messages' in chunk[ 'tools' ]:
for message in chunk[ 'tools' ][ 'messages' ]:
if hasattr (message, 'name' ) and hasattr (message, 'content' ):
# 顯示工具響應
display(Markdown( f"**Tool ( {message.name} )**: {message.content} " ))運行此代碼將逐行輸出內容：
參考文獻
譯者介紹
朱先忠，51CTO社區編輯，51CTO專家博客、講師，濰坊一所高校計算機教師，自由編程界老兵一枚。
原文標題：Model Context Protocol With LangChain Agent Client，作者：Prince Krampah
