譯者 | 朱先忠
審校 | 重樓
簡介
什么是模型上下文協(xié)議（Model Context Protocol）？讓我們深入了解MCP背后的概念。以下是官方MCP文檔對MCP的介紹：
“MCP是一種開放協(xié)議，它標準化了應(yīng)用程序向LLM提供上下文的方式?？梢詫CP視為AI應(yīng)用程序的USB-C端口。正如USB-C提供了一種將你的設(shè)備連接到各種外圍設(shè)備和配件的標準化方式一樣，MCP提供了一種將AI模型連接到不同數(shù)據(jù)源和工具的標準化方式?！?/p>
讓我來解釋一下。假設(shè)你正在構(gòu)建與不同語言模型和AI系統(tǒng)配合使用的AI代理，其中每個模型對工具的理解方式都不同。你已經(jīng)編寫了代碼來使你的AI代理能夠針對一個特定的AI模型進行構(gòu)建，假設(shè)你的系統(tǒng)構(gòu)架如下圖所示：
假設(shè)你將來可能希望切換到具有不同架構(gòu)和工具定義方法的另一個AI模型，那么你必須回去重新編寫工具以適應(yīng)這種新的AI模型架構(gòu)和方法，假設(shè)新架構(gòu)如下圖所示：
我想，作為程序員，你已經(jīng)看到了這里的問題——這是不可擴展的。如果我們可以編寫一次工具，然后能夠?qū)⑵渑c任何AI模型架構(gòu)連接起來，而不必擔(dān)心這個AI模型架構(gòu)在后臺如何工作，那會怎樣呢？
這會為我們省去很多麻煩，不是嗎？是的，它不僅可擴展，我們還可以連接任何我們想要的AI模型！
為什么選擇MCP？
你可能會想，我們剛剛引入了另一層（MCP層），另一層就意味著更復(fù)雜嗎？是的，但增加這一層的好處遠遠大于壞處。下面是官方文檔的說明：
“MCP可幫助你在LLM之上構(gòu)建代理和復(fù)雜的工作流。LLM通常需要與數(shù)據(jù)和工具集成，而MCP可提供：
- 你的LLM可以直接插入不斷增加的預(yù)構(gòu)建集成列表
- 在LLM提供商和供應(yīng)商之間切換的靈活性
- 保護基礎(chǔ)架構(gòu)內(nèi)數(shù)據(jù)的最佳實踐”
MCP的總體架構(gòu)
總體來說，MCP架構(gòu)遵循客戶端-服務(wù)器架構(gòu)。我們可以讓一個客戶端連接到多個服務(wù)器（MCP服務(wù)器）。
現(xiàn)在，讓我們來分析一下上面的圖形架構(gòu)：
- MCP主機：頂部的“主機（代理、工具）”框代表想要通過模型上下文協(xié)議訪問數(shù)據(jù)的程序。
- MCP客戶端：通過MCP協(xié)議直接與MCP服務(wù)器（A、B、C）連接的客戶端。
- MCP服務(wù)器：用三個框表示（MCP服務(wù)器A、B、C），每個框連接到不同的服務(wù)。
- 本地數(shù)據(jù)源：文件系統(tǒng)和本地Postgres數(shù)據(jù)庫。
- 遠程服務(wù)：虛擬私有云之外的Postgres存儲。該架構(gòu)顯示了一個包含MCP基礎(chǔ)設(shè)施的VPC（虛擬私有云）。其中，主機與多個MCP服務(wù)器通信，每個服務(wù)器處理特定的服務(wù)集成。
MCP中的核心概念
- 資源：客戶端可以訪問和讀取的數(shù)據(jù)對象（類似于文件或API響應(yīng)）。
- 工具：LLM可以觸發(fā)的可執(zhí)行函數(shù)（需要用戶權(quán)限）。
- 提示：現(xiàn)成的文本模板，旨在幫助用戶完成特定任務(wù)。
創(chuàng)建你的第一個MCP服務(wù)器
對于我們的第一個MCP服務(wù)器，我想直接在官方文檔上創(chuàng)建一個天氣MCP服務(wù)器，這只是為了讓我們更迅速地了解MCP的方式。然后，我們將此服務(wù)器連接到一個LangChain代理。
如果愿意，你可以按照官方文檔中的說明進行操作，我也會在本文中提供相關(guān)步驟。
安裝環(huán)境
在本文案例中，我們選擇使用uv包管理器，它是推薦的包管理器，而且速度非?？欤晕視猿质褂盟?。通過運行下面的命令來安裝它：
curl -LsSf https://astral.sh/uv/install.sh | sh
至此，我已經(jīng)成功地將它安裝在我的機器上。如果這是你第一次安裝它，你可能需要重新啟動你的終端。
簡言之，我正在使用Ma/Linux命令。如果你使用的是Windows，你可以按照官方文檔中的Powershell命令進行操作。
# 為我們的項目創(chuàng)建一個新目錄
uv init weather
cd weather
# 創(chuàng)建虛擬環(huán)境并激活它
uv venv
source .venv/bin/activate
# 安裝依賴項
uv add "mcp[cli]" httpx
# 創(chuàng)建我們的服務(wù)器文件
touch weather.py
完成后，你可以在你最喜歡的IDE中打開目錄。我將使用VSCode；如果你愿意，也可以使用Cursor或者任何其他IDE。
code .編寫服務(wù)器端代碼
對于本文中的代碼，我將使用官方文檔中的代碼。在此，非常感謝MCP團隊提供的代碼。
首先，我們將實例化FastMCP類，這有助于大多數(shù)工具創(chuàng)建邏輯，例如來自工具函數(shù)的文檔字符串的工具描述以及函數(shù)類型提示。
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
#初始化FastMCP服務(wù)器
mcp = FastMCP("weather")
# 指定常量
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"輔助函數(shù)
我們還將創(chuàng)建幾個輔助函數(shù)，用于幫助格式化來自API的數(shù)據(jù)。
async def make_nws_request(url: str) -> dict[str, Any] | None:
"""使用恰當?shù)腻e誤處理方式向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')}
"""創(chuàng)建工具
現(xiàn)在，我們將使用Python中的裝飾器在MCP服務(wù)器下創(chuàng)建實際的工具mcp.tool()。
@mcp.tool()
async def get_alerts(state: str) -> str:
"""獲取美國一個州的天氣警報。
參數(shù):
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:
"""獲取一個地點的天氣預(yù)報。
參數(shù):
latitude: 位置的緯度
longitude: 位置的經(jīng)度
"""
# 首先獲取預(yù)測網(wǎng)格端點
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."
# 從端點響應(yīng)中獲取預(yù)測的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."
# 將時間范圍格式化為可讀的預(yù)測
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)啟動服務(wù)器
一旦我們完成上面所有這些工作，我們就可以在腳本中添加入口點來執(zhí)行MCP服務(wù)器。現(xiàn)在，在腳本文件weather.py的底部添加以下代碼：
if __name__ == "__main__" :
# 初始化并運行服務(wù)器
mcp.run(transport= 'stdio' )從上面的代碼中我們指定了stdio，這是什么意思？
HTTP中的STDIO（標準輸入/輸出）是指使用HTTP連接時輸入和輸出數(shù)據(jù)的標準流。在Web服務(wù)器和HTTP環(huán)境中：
- 標準輸入（stdin）：用于接收發(fā)送到服務(wù)器的數(shù)據(jù)，如POST請求數(shù)據(jù)。
- 標準輸出（stdout）：用于將響應(yīng)數(shù)據(jù)發(fā)送回客戶端。
- 標準錯誤（stderr）：用于記錄錯誤和調(diào)試信息在構(gòu)建使用命令行界面的HTTP服務(wù)器或服務(wù)時，STDIO提供了一種通過標準Unix風(fēng)格流傳輸HTTP請求/響應(yīng)數(shù)據(jù)的方法，允許與其他命令行工具和進程集成。
我們還可以指定SSE通信方式： - HTTP技術(shù)允許服務(wù)器將更新推送到客戶端。
- 單向通信（僅限服務(wù)器到客戶端）。
- 保持連接暢通以獲取實時更新。
- 比WebSocket更簡單。
- 用于通知、數(shù)據(jù)饋送和流更新。
完成后，導(dǎo)航到weather.py腳本所在的位置并在終端中運行以下命令：
uv run weather.py
除了看不到輸出內(nèi)容之外，這表明服務(wù)器正在運行，或者你可以更新腳本以顯示某些內(nèi)容（如果你愿意）。
連接到客戶端
你可以使用不同的客戶端連接到此服務(wù)器，例如Claude桌面客戶端、Cursor和許多其他客戶端。你可以在此處閱讀更多相關(guān)信息。
LangChain代理MCP客戶端
我想創(chuàng)建一個自定義LangChain代理來連接我們正在運行的MCP服務(wù)器。為此，我們必須安裝langchain-mcp-adapters。你可以運行以下命令。
首先，停止天氣腳本并運行以下命令：
uv add ipykernel原因是我將在VScode中使用一個筆記本文件作為LangChain代理。
安裝完成后，繼續(xù)再次運行天氣MCP服務(wù)器腳本：
uv run weather.py我還在與我們的文件weather.py相同的目錄中繼續(xù)創(chuàng)建另一個文件client.ipynb。
然后，你可以運行下面的命令來安裝LangChain MCP適配器：
!uv add langchain-mcp-adapters安裝完成后，我們可以安裝langchain-anthropic和LangGraph客戶端。
!uv add langgraph langchain-anthropic python-dotenv
加載環(huán)境變量
首先，我們需要一個Anthropic API密鑰。
一旦你獲得Anthropic API密鑰，你就可以添加.env文件，該文件應(yīng)該位于你的項目的根目錄中。
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連接創(chuàng)建服務(wù)器參數(shù)
現(xiàn)在，我們可以創(chuàng)建與我們正在運行的MCP服務(wù)器的stdio連接服務(wù)器參數(shù)。
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讓我們繼續(xù)創(chuàng)建模型，我將使用前面提到的Anthropic。我們再次提及了如何使用多個LLM提供商的問題。
model = ChatAnthropic(model="claude-3-5-sonnet-20241022", api_key=api_key)你可以在下面的鏈接處找到有關(guān)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)
# 創(chuàng)建并運行代理
agent = create_react_agent(model, tools)
agent_response = await agent.ainvoke({ "messages" : "加州目前的天氣怎么樣" })然后，運行下面命令：
agent_response
漂亮的輸出
現(xiàn)在，讓我們讓輸出看起來更美觀一點：
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協(xié)議上運行MCP（響應(yīng)流）
我希望能夠流式傳輸響應(yīng)，為此我們需要將trasportMCP的類型設(shè)置為sse。
為此，停止服務(wù)器（MCP服務(wù)器）并更改這部分代碼：
if __name__ == "__main__" :
# 初始化并運行服務(wù)器
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}"))
流式響應(yīng)
我希望能夠?qū)崟r流式傳輸響應(yīng)。為此，讓我們編寫以下代碼行：
async with MultiServerMCPClient(
{
"weather" : {
"url" : "http://localhost:8000/sse" ,
"transport" : "sse" ,
}
}
) as client:
agent = create_react_agent(model, client.get_tools())
# 流式傳輸響應(yīng)塊
async for chunk in agent.astream({ "messages" : "what is the weather in nyc!" }):
# 從AddableUpdatesDict結(jié)構(gòu)中提取消息內(nèi)容
if 'agent' in chunk and 'messages' in chunk[ 'agent' ]:
for message in chunk[ 'agent' ][ 'messages' ]:
if isinstance (message, AIMessage):
# 處理不同的內(nèi)容格式
if isinstance (message.content, list ):
# 對于帶有文本和工具使用的結(jié)構(gòu)化內(nèi)容
for item in message.content:
if isinstance (item, dict ) and 'text' in item:
display(Markdown( f"**AI**: {item[ 'text' ]} " ))
else :
# 對于簡單文本內(nèi)容
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' ):
# 顯示工具響應(yīng)
display(Markdown( f"**Tool ( {message.name} )**: {message.content} " ))運行此代碼將逐行輸出內(nèi)容：
參考文獻
譯者介紹
朱先忠，51CTO社區(qū)編輯，51CTO專家博客、講師，濰坊一所高校計算機教師，自由編程界老兵一枚。
原文標題：Model Context Protocol With LangChain Agent Client，作者：Prince Krampah
