如何使用LangChain建立MCP客戶端伺服器

如何使用LangChain建立MCP客戶端伺服器

人工智慧和大型語言模型 (LLM) 的世界瞬息萬變。整合外部工具和即時資料對於構建真正強大的應用程式至關重要。模型上下文協議(MCP)提供了彌合這一差距的標準方法。本教學指南為使用 LangChain 建立 MCP 客戶端伺服器提供了清晰易懂的新手指南。瞭解 MCP 客戶端伺服器架構有助於構建強大的人工智慧代理。我們將介紹一些基本知識,包括什麼是 MCP 伺服器功能,並提供一個使用 LangChain 的實用 MCP 客戶端伺服器示例。

瞭解模型上下文協議(MCP)

模型上下文協議(MCP)

Source: Link

那麼,MCP 伺服器和客戶端互動是怎麼回事?模型上下文協議(MCP)是一個開放標準系統。Anthropic 開發該系統是為了將 LLM 與外部工具和資料來源有效連線起來。它採用結構化和可重複使用的方法。MCP 可幫助人工智慧模型與不同系統對話。這使它們能夠訪問當前資訊,並完成超出其初始訓練的任務。將其視為人工智慧與外部世界之間的通用翻譯器,構成了 MCP 客戶端伺服器架構的核心。

MCP的主要特點

MCP 具有以下幾個重要特點:

  1. 標準化整合:MCP 提供了一種單一、一致的方式,將 LLM 與許多工具和資料來源連線起來。這樣就不需要為每個連線編寫獨特的程式碼。它簡化了使用 LangChain 設定 MCP 客戶端伺服器的過程。
  2. 上下文管理:該協議可確保人工智慧模型在多個步驟中跟蹤對話上下文。這可以防止在任務需要多次互動時丟失重要資訊。
  3. 安全和隔離:MCP 包括強大的安全措施。它嚴格控制訪問許可權,並使用許可權邊界將伺服器連線分開。這確保了客戶端和伺服器之間的安全通訊。

MCP在基於LLM的應用中的作用

LLM 應用程式經常需要外部資料。它們可能需要查詢資料庫、獲取文件或使用網路 API。MCP 是一個重要的中間層。它能讓模型與這些外部資源順利互動,而無需手動操作。使用 LangChain 的 MCP 客戶端伺服器,開發人員可以構建更智慧的人工智慧代理。這些代理能力更強、工作速度更快,並能在定義明確的 MCP 客戶端伺服器架構內安全執行。這種設定是高階人工智慧助手的基礎。現在,讓我們來看看實施部分。

設定環境

在使用 LangChain 構建 MCP 客戶端伺服器之前,我們先來準備一下環境。您需要這些物品:

  • Python 3.11 或更新版本。
  • 設定新的虛擬環境(可選)
  • 一個 API 金鑰(例如 OpenAI 或 Groq,取決於您選擇的模式)。
  • 特定的 Python 庫:langchain-mcp-adapterslanggraph 和您選擇的 LLM 庫(如 langchain-openai langchain groq)。

使用 pip 安裝所需的庫。開啟終端或命令提示符並執行:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
pip install langchain-mcp-adapters langgraph langchain-groq # Or langchain-openai
pip install langchain-mcp-adapters langgraph langchain-groq # Or langchain-openai
pip install langchain-mcp-adapters langgraph langchain-groq # Or langchain-openai

確保已準備好正確的 Python 版本和必要的金鑰。

構建MCP伺服器

MCP 伺服器的任務是提供客戶端可以使用的工具。在使用 langchain 的 MCP 客戶端伺服器示例中,我們將構建一個簡單的伺服器。該伺服器將處理基本的數學運算以及複雜的天氣 api,以獲取城市的天氣詳細資訊。從這裡開始瞭解 MCP 伺服器的功能。

建立一個名為 mcp_server.py 的 Python 檔案:

1. 讓我們匯入所需的庫

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
import math
import requests
from mcp.server.fastmcp import FastMCP
import math import requests from mcp.server.fastmcp import FastMCP
import math
import requests
from mcp.server.fastmcp import FastMCP

2. 初始化 FastMCP 物件

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
mcp= FastMCP("Math")
mcp= FastMCP("Math")
mcp= FastMCP("Math")

3. 讓我們來定義數學工具

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
@mcp.tool()
def add(a: int, b: int) -> int:
print(f"Server received add request: {a}, {b}")
return a + b
@mcp.tool()
def multiply(a: int, b: int) -> int:
print(f"Server received multiply request: {a}, {b}")
return a * b
@mcp.tool()
def sine(a: int) -> int:
print(f"Server received sine request: {a}")
return math.sin(a)
@mcp.tool() def add(a: int, b: int) -> int: print(f"Server received add request: {a}, {b}") return a + b @mcp.tool() def multiply(a: int, b: int) -> int: print(f"Server received multiply request: {a}, {b}") return a * b @mcp.tool() def sine(a: int) -> int: print(f"Server received sine request: {a}") return math.sin(a)
@mcp.tool()
def add(a: int, b: int) -> int:
   print(f"Server received add request: {a}, {b}")
   return a + b
@mcp.tool()
def multiply(a: int, b: int) -> int:
   print(f"Server received multiply request: {a}, {b}")
   return a * b
@mcp.tool()
def sine(a: int) -> int:
   print(f"Server received sine request: {a}")
   return math.sin(a)

4. 現在,讓我們定義一個天氣工具,確保您已從這裡獲取了 API。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
WEATHER_API_KEY = "YOUR_API_KEY"
@mcp.tool()
def get_weather(city: str) -> dict:
"""
Fetch current weather for a given city using WeatherAPI.com.
Returns a dictionary with city, temperature (C), and condition.
"""
print(f"Server received weather request: {city}")
url = f"http://api.weatherapi.com/v1/current.json?key={WEATHER_API_KEY}&q={city}"
response = requests.get(url)
if response.status_code != 200:
return {"error": f"Failed to fetch weather for {city}."}
data = response.json()
return {
"city": data["location"]["name"],
"region": data["location"]["region"],
"country": data["location"]["country"],
"temperature_C": data["current"]["temp_c"],
"condition": data["current"]["condition"]["text"]
}
5. Now, instantiate the mcp server
if __name__ =="__main__":
print("Starting MCP Server....")
mcp.run(transport="stdio")
WEATHER_API_KEY = "YOUR_API_KEY" @mcp.tool() def get_weather(city: str) -> dict: """ Fetch current weather for a given city using WeatherAPI.com. Returns a dictionary with city, temperature (C), and condition. """ print(f"Server received weather request: {city}") url = f"http://api.weatherapi.com/v1/current.json?key={WEATHER_API_KEY}&q={city}" response = requests.get(url) if response.status_code != 200: return {"error": f"Failed to fetch weather for {city}."} data = response.json() return { "city": data["location"]["name"], "region": data["location"]["region"], "country": data["location"]["country"], "temperature_C": data["current"]["temp_c"], "condition": data["current"]["condition"]["text"] } 5. Now, instantiate the mcp server if __name__ =="__main__": print("Starting MCP Server....") mcp.run(transport="stdio")
WEATHER_API_KEY = "YOUR_API_KEY"
@mcp.tool()
def get_weather(city: str) -> dict:
   """
   Fetch current weather for a given city using WeatherAPI.com.
   Returns a dictionary with city, temperature (C), and condition.
   """
   print(f"Server received weather request: {city}")
   url = f"http://api.weatherapi.com/v1/current.json?key={WEATHER_API_KEY}&q={city}"
   response = requests.get(url)
   if response.status_code != 200:
       return {"error": f"Failed to fetch weather for {city}."}
   data = response.json()
   return {
       "city": data["location"]["name"],
       "region": data["location"]["region"],
       "country": data["location"]["country"],
       "temperature_C": data["current"]["temp_c"],
       "condition": data["current"]["condition"]["text"]
   }
   5. Now, instantiate the mcp server 
if __name__ =="__main__":
   print("Starting MCP Server....")
   mcp.run(transport="stdio")

說明:

該指令碼設定了一個名為“Math”的簡單 MCP 伺服器。它使用 FastMCP 定義了四個工具:加法、乘法、正弦和由 @mcp.tool() 裝飾器標記的 get_weather 。型別提示告訴 MCP 有關預期輸入和輸出的資訊。直接執行時,伺服器使用標準輸入/輸出(stdio)進行通訊。以上演示了 MCP 伺服器的基本設定。

執行伺服器:開啟終端,導航到包含 mcp_server.py 的目錄。然後執行

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
python mcp_server.py
python mcp_server.py
python mcp_server.py

伺服器啟動時應無任何警告。該伺服器將持續執行,以便客戶端訪問工具

輸出:

mcp伺服器輸出

構建MCP客戶端

客戶端連線到伺服器,傳送請求(如要求代理執行計算並獲取即時天氣),並處理響應。這演示了使用 LangChain 的 MCP 客戶端伺服器的客戶端。

建立一個名為 client.py 的 Python 檔案:

  1. 首先匯入必要的庫
Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
# client.py
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_groq import ChatGroq
from langchain_openai import ChatOpenAI
import asyncio
import os
# client.py 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_groq import ChatGroq from langchain_openai import ChatOpenAI import asyncio import os
# client.py
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_groq import ChatGroq
from langchain_openai import ChatOpenAI
import asyncio
import os
  1. 為 LLM(Groq 或 OpenAI)設定 API 金鑰,並初始化 LLM 模型
Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
# Set your API key (replace with your actual key or use environment variables)
GROQ_API_KEY = "YOUR_GROQ_API_KEY" # Replace with your key
os.environ["GROQ_API_KEY"] = GROQ_API_KEY
# OPENAI_API_KEY = "YOUR_OPENAI_API_KEY"
# os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY
# Initialize the LLM model
model = ChatGroq(model="llama3-8b-8192", temperature=0)
# model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# Set your API key (replace with your actual key or use environment variables) GROQ_API_KEY = "YOUR_GROQ_API_KEY" # Replace with your key os.environ["GROQ_API_KEY"] = GROQ_API_KEY # OPENAI_API_KEY = "YOUR_OPENAI_API_KEY" # os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY # Initialize the LLM model model = ChatGroq(model="llama3-8b-8192", temperature=0) # model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# Set your API key (replace with your actual key or use environment variables)
GROQ_API_KEY = "YOUR_GROQ_API_KEY" # Replace with your key
os.environ["GROQ_API_KEY"] = GROQ_API_KEY
# OPENAI_API_KEY = "YOUR_OPENAI_API_KEY"
# os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY
# Initialize the LLM model
model = ChatGroq(model="llama3-8b-8192", temperature=0)
# model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
  1. 現在,定義引數以啟動 MCP 伺服器程序。
Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
server_params = StdioServerParameters(
command="python", # Command to execute
args=["mcp_server.py"] # Arguments for the command (our server script)
)
server_params = StdioServerParameters( command="python", # Command to execute args=["mcp_server.py"] # Arguments for the command (our server script) )
server_params = StdioServerParameters(
   command="python",      # Command to execute
   args=["mcp_server.py"] # Arguments for the command (our server script)
)
  1. 讓我們定義執行代理互動的非同步函式
Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
async def run_agent():
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
print("MCP Session Initialized.")
tools = await load_mcp_tools(session)
print(f"Loaded Tools: {[tool.name for tool in tools]}")
agent = create_react_agent(model, tools)
print("ReAct Agent Created.")
print(f"Invoking agent with query")
response = await agent.ainvoke({
"messages": [("user", "What is (7+9)x17, then give me sine of the output recieved and then tell me What's the weather in Torronto, Canada?")]
})
print("Agent invocation complete.")
# Return the content of the last message (usually the agent's final answer)
return response["messages"][-1].content
async def run_agent(): async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() print("MCP Session Initialized.") tools = await load_mcp_tools(session) print(f"Loaded Tools: {[tool.name for tool in tools]}") agent = create_react_agent(model, tools) print("ReAct Agent Created.") print(f"Invoking agent with query") response = await agent.ainvoke({ "messages": [("user", "What is (7+9)x17, then give me sine of the output recieved and then tell me What's the weather in Torronto, Canada?")] }) print("Agent invocation complete.") # Return the content of the last message (usually the agent's final answer) return response["messages"][-1].content
async def run_agent():
   async with stdio_client(server_params) as (read, write):
       async with ClientSession(read, write) as session:
           await session.initialize()
           print("MCP Session Initialized.")
           tools = await load_mcp_tools(session)
           print(f"Loaded Tools: {[tool.name for tool in tools]}")
           agent = create_react_agent(model, tools)
           print("ReAct Agent Created.")
           print(f"Invoking agent with query")
           response = await agent.ainvoke({
               "messages": [("user", "What is (7+9)x17, then give me sine of the output recieved and then tell me What's the weather in Torronto, Canada?")]
           })
           print("Agent invocation complete.")
           # Return the content of the last message (usually the agent's final answer)
           return response["messages"][-1].content
  1. 現在,執行此函式並在終端上等待結果
Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
# Standard Python entry point check
if __name__ == "__main__":
# Run the asynchronous run_agent function and wait for the result
print("Starting MCP Client...")
result = asyncio.run(run_agent())
print("\nAgent Final Response:")
print(result)
# Standard Python entry point check if __name__ == "__main__": # Run the asynchronous run_agent function and wait for the result print("Starting MCP Client...") result = asyncio.run(run_agent()) print("\nAgent Final Response:") print(result)
# Standard Python entry point check
if __name__ == "__main__":
   # Run the asynchronous run_agent function and wait for the result
   print("Starting MCP Client...")
   result = asyncio.run(run_agent())
   print("\nAgent Final Response:")
   print(result)

說明:

該客戶端指令碼用於配置 LLM(此處使用 ChatGroq;記得設定 API 金鑰)。它使用 StdioServerParameters 定義瞭如何啟動伺服器。run_agent 函式透過 stdio_client 連線到伺服器,建立一個 ClientSession 並將其初始化。 load_mcp_tool 為 LangChain 獲取伺服器工具。create_react_agent 使用 LLM 和工具處理使用者查詢。最後,agent.ainvoke 傳送查詢,讓代理可能使用伺服器工具找到答案。下面是一個使用 langchain 的完整 MCP 客戶端伺服器示例。

執行客戶端:

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
python client.py
python client.py
python client.py

輸出:

mcp客戶端輸出

我們可以看到,客戶端啟動伺服器程序、初始化連線、載入工具、呼叫代理並列印透過呼叫伺服器的新增工具(也稱為天氣 api)和檢索即時天氣資料計算出的最終答案。

實際應用

使用 LangChain 的 MCP 客戶端伺服器為建立複雜的人工智慧代理提供了多種可能性。一些實際應用包括

  • LLM 獨立性:透過使用 Langchain,我們現在可以將任何 LLM 與 MCP 整合。以前我們只能
  • 資料檢索:代理可以透過 MCP 連線到資料庫伺服器,獲取即時客戶資料或查詢內部知識庫。
  • 文件處理:代理可以使用 MCP 工具與文件管理系統進行互動,從而可以根據使用者請求彙總、提取資訊或更新文件。
  • 任務自動化:透過 MCP 伺服器與各種業務系統(如客戶關係管理系統、日曆或專案管理工具)整合,自動執行日常任務,如安排會議或更新銷售記錄。MCP 客戶端伺服器架構支援這些複雜的工作流程。

最佳實踐

在使用 LangChain 構建 MCP 客戶端伺服器時,請遵循良好實踐,以獲得更好的效果:

  • 採用模組化設計,為不同的任務建立特定的工具,並將伺服器邏輯與客戶端邏輯分開。
  • 在伺服器工具和客戶端代理中實施強大的錯誤處理功能,以便系統能從容應對故障。
  • 透過使用 MCP 的訪問控制和許可權邊界等功能,優先考慮安全性,尤其是當伺服器處理敏感資料時。
  • 為 MCP 工具提供清晰的說明和文件;這有助於代理瞭解其目的和用法。

常見陷阱

開發系統時要注意潛在問題。如果代理框架不能正確管理狀態,複雜對話中可能會出現上下文丟失,從而導致錯誤。長期執行的 MCP 伺服器資源管理不善可能會導致記憶體洩漏或效能下降,因此要謹慎處理連線和檔案控制代碼。確保客戶端和伺服器傳輸機制的相容性,因為不匹配(如一個使用 stdio 而另一個期待 HTTP)會阻礙通訊。最後,注意工具模式的不匹配,伺服器工具的定義與客戶端的期望不一致時,會阻礙工具的執行。解決這些問題可以加強使用 LangChain 實現的 MCP 客戶端伺服器。

小結

利用 LangChain 的模型上下文協議為構建高階人工智慧代理提供了一種功能強大的標準化方法。透過使用 LangChain 建立 MCP 客戶端伺服器,您可以讓 LLM 安全有效地與外部工具和資料來源進行互動。本指南使用 LangChain 示例了一個基本的 MCP 客戶端伺服器,概述了 MCP 客戶端伺服器的核心架構以及 MCP 伺服器的功能。這種方法可簡化整合、提高代理能力並確保可靠執行,從而為更智慧、更有用的人工智慧應用鋪平道路。

評論留言