然后,統一客戶端和服務端的運行規范.并且要求MCP客戶端和服務端之間,也統一按照某個既定的提示詞模板進行通信.
這個的思路很簡單就不多介紹了，主要是function calling有什么問題?function calling最大的問題就在于它的工具量非常大，一個簡單的外部函數往往就要上百行的代碼，并且，為了讓大模型認識這些外部函數，還需要為每個外部函數編寫一個json schema格式的功能說明，此外還要精心設計一個提示詞模板才能提高function calling響應的準確率。
所以MCP的目標，就是為了簡化這一過程。先統一名稱。把大模型的運行環境成為MCP Clinet，也就是MCP客戶端，同時，把外部函數運行環境作為MCP Server，MCP服務端。
然后,統一客戶端和服務端的運行規范.并且要求MCP客戶端和服務端之間,也統一按照某個既定的提示詞模板進行通信.
這樣做目前最大的好處就在于MCP服務器的開發成為一個特定的模板,并且只要開發了一個MCP server之后,大家也就都能夠直接進行使用.比如查詢天氣,網頁爬取,查詢本地SQL等等通用需求.
于是github中的MCP成堆出現,大模型能夠使用的工具不斷增多,
關于agent的入門學習理論，主要包含的四個模塊及其作用方面.
我看大模型使用的最多的環境是uv的虛擬環境,這里稍微講一下他們之間的關系.conda是一個跨語言的包和環境的管理工具(不止有python環境),uv則是一個超快的,主要用于python生態,和pip poetry是一類的
這里我們選擇uv來構建,也借著學習一下uv的安裝和命令,首先是通過pip來安裝uv,python的版本的話優先考慮py3.11的.
pip install uv安裝依賴庫
uv pip install requests
# 與pip install 類似,但是更快創建虛擬環境.到這里才是真正使用uv創建虛擬環境:
uv venv myenv這里不像conda,可以指定自己的python版本,這里默認就是使用當前python版本進行虛擬的.類似于運行了:python -m venv myenv.
然后就可以激活環境:
source myenv/bin/activatewindows中則是:
myenvScriptsactivate
# Windows然后在虛擬環境中都要通過uv開頭的命令來運行文件,這樣能夠更加高效.
上面主要是uv的一些基礎內容,接下來我們正式建立mcp client和mcp server
創建mcp客戶端:
我用的是自己在win上安裝的py3.8
uv init mcp-client
cd mcp-client
然后創建虛擬環境并進入:
uv venv
.venvScriptsactivate
接下來安裝一些依賴庫:
uv add mcp openai python-dotenv報錯了:
因為mcp 包要求 Python 版本大于等于 3.10，這就導致依賴解析失敗。所以我們推薦的還是py3.11的版本.
那怎么通過uv創建不同的python環境呢?
uv沒有這樣的功能.所以說我們還是需要一個conda,創建一個py3.11的版本,然后重新直線上述過程.還挺無語的,但是要想要uv的快速原型功能也只好這樣了…
conda create -n mcp_server python=3.11
conda activate mcp_server然后重新uv:
uv init mcp-client
cd mcp-client
uv venv
.venvScriptsactivate
uv add mcp openai python-dotenv
python雙重環境…
安裝完畢之后:
創建客戶端client.py,是調用大模型的核心部分:
if name == "main":
asyncio.run(main())
# 加載 .env 文件，確保 API Key 受到保護
load_dotenv()
class MCPClient:
def __init__(self):
"""初始化 MCP 客戶端"""
self.exit_stack = AsyncExitStack()
self.openai_api_key = os.getenv("OPENAI_API_KEY")
# 讀取 OpenAI API Key
self.base_url = os.getenv("BASE_URL")
# 讀取 BASE YRL
self.model = os.getenv("MODEL")
# 讀取 model
if not self.openai_api_key:
raise ValueError("? 未找到 [OpenAI](http://www.dlbhg.com/provider/uid202405280134182e2ed4) [API](http://www.dlbhg.com/wiki/api/) Key，請在 .env 文件中設置 OPENAI_API_KEY")
self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)
async def process_query(self, query: str) -> str:
"""[調用 OpenAI API](http://www.dlbhg.com/blog/ua-using-pycharm-to-call-api-guide) 處理用戶查詢"""
messages = [{"role": "system", "content": "你是一個智能助手，幫助用戶回答問題。"},
{"role": "user", "content": query}]
try:
# 調用 OpenAI API
response = await asyncio.get_event_loop().run_in_executor(
None,
lambda: self.client.chat.completions.create(
model=self.model,
messages=messages,
stream=False
)
)
return response.choices[0].message.content
except Exception as e:
return f"?? 調用 OpenAI API 時出錯: {str(e)}"
async def chat_loop(self):
"""運行交互式聊天循環"""
print("n?? MCP 客戶端已啟動！輸入 'quit' 退出")
while True:
try:
query = input("n你: ").strip()
if query.lower() == 'quit':
break
response = await self.process_query(query)
# 發送用戶輸入到 OpenAI API
print(f"n?? OpenAI: {response}")
except Exception as e:
print(f"n?? 發生錯誤: {str(e)}")
async def cleanup(self):
"""清理資源"""
await self.exit_stack.aclose()
async def main():
client = MCPClient()
try:
await client.chat_loop()
finally:
await client.cleanup()
if __name__ == "__main__":
asyncio.run(main())然后在文件夾中新建.env文件,寫入對應的api-key:
我測試過了,deepseek官網的api才支持后續的工具的使用,或者是昂貴的openai和claude,這里就不考慮了
BASE_URL=https://api.deepseek.com
MODEL=deepseek-chat
OPENAI_API_KEY="sk-xxx"運行:
uv run client.py
然后是服務端的核心部分,我們以天氣調用的工具為例:
首先你需要一個openweather的API來查詢天氣:https://openweathermap.org/api
注冊一個賬號之后去生成一個API即可
服務端的依賴項安裝:
uv add mcp httpx新建一個server.py:
mcp.run(transport='stdio')# 初始化 MCP 服務器
mcp = FastMCP("WeatherServer")
# OpenWeather API 配置
OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
API_KEY = "xxx"
# 請替換為你自己的 OpenWeather API Key
USER_AGENT = "weather-app/1.0"
async def fetch_weather(city: str) -> dict[str, Any] | None:
"""
從 OpenWeather API 獲取天氣信息。
:param city: 城市名稱（需使用英文，如 Beijing）
:return: 天氣數據字典；若出錯返回包含 error 信息的字典
"""
params = {
"q": city,
"appid": API_KEY,
"units": "metric",
"lang": "zh_cn"
}
headers = {"User-[Agent](http://www.dlbhg.com/wiki/what-is-an-ai-agent/)": USER_AGENT}
async with httpx.AsyncClient() as client:
try:
response = await client.get(OPENWEATHER_API_BASE, params=params, headers=headers, timeout=30.0)
response.raise_for_status()
return response.json()
# 返回字典類型
except httpx.HTTPStatusError as e:
return {"error": f"HTTP 錯誤: {e.response.status_code}"}
except Exception as e:
return {"error": f"請求失敗: {str(e)}"}
def format_weather(data: dict[str, Any] | str) -> str:
"""
將天氣數據格式化為易讀文本。
:param data: 天氣數據（可以是字典或 JSON 字符串）
:return: 格式化后的天氣信息字符串
"""
# 如果傳入的是字符串，則先轉換為字典
if isinstance(data, str):
try:
data = json.loads(data)
except Exception as e:
return f"無法解析天氣數據: {e}"
# 如果數據中包含錯誤信息，直接返回錯誤提示
if"error"in data:
return f"?? {data['error']}"
# 提取數據時做容錯處理
city = data.get("name", "未知")
country = data.get("sys", {}).get("country", "未知")
temp = data.get("main", {}).get("temp", "N/A")
humidity = data.get("main", {}).get("humidity", "N/A")
wind_speed = data.get("wind", {}).get("speed", "N/A")
# weather 可能為空列表，因此用 [0] 前先提供默認字典
weather_list = data.get("weather", [{}])
description = weather_list[0].get("description", "未知")
return (
f"?? {city}, {country}n"
f"?? 溫度: {temp}°Cn"
f"?? 濕度: {humidity}%n"
f"?? 風速: {wind_speed} m/sn"
f"?? 天氣: {description}n"
)
@mcp.tool()
async def query_weather(city: str) -> str:
"""
輸入指定城市的英文名稱，返回今日天氣查詢結果。
:param city: 城市名稱（需使用英文）
:return: 格式化后的天氣信息
"""
data = await fetch_weather(city)
return format_weather(data)
if __name__ == "__main__":
# 以標準 I/O 方式運行 MCP 服務器
mcp.run(transport='stdio')新的客戶端部分client_new.py:
if name == "main":
import sys
asyncio.run(main())
from openai import OpenAI
from dotenv import load_dotenv
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
# 加載 .env 文件，確保 API Key 受到保護
load_dotenv()
class MCPClient:
def __init__(self):
"""初始化 MCP 客戶端"""
self.exit_stack = AsyncExitStack()
self.openai_api_key = os.getenv("OPENAI_API_KEY")
# 讀取 OpenAI API Key
self.base_url = os.getenv("BASE_URL")
# 讀取 BASE YRL
self.model = os.getenv("MODEL")
# 讀取 model
if not self.openai_api_key:
raise ValueError("? 未找到 OpenAI API Key，請在 .env 文件中設置 OPENAI_API_KEY")
self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)
# 創建OpenAI client
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()
async def connect_to_server(self, server_script_path: str):
"""連接到 MCP 服務器并列出可用工具"""
is_python = server_script_path.endswith('.py')
is_js = server_script_path.endswith('.js')
if not (is_python or is_js):
raise ValueError("服務器腳本必須是 .py 或 .js 文件")
command = "python"if is_python else"node"
server_params = StdioServerParameters(
command=command,
args=[server_script_path],
env=None
)
# 啟動 MCP 服務器并建立通信
stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
self.stdio, self.write = stdio_transport
self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
await self.session.initialize()
# 列出 MCP 服務器上的工具
response = await self.session.list_tools()
tools = response.tools
print("n已連接到服務器，支持以下工具:", [tool.name for tool in tools])
async def process_query(self, query: str) -> str:
"""
使用大模型處理查詢并調用可用的 MCP 工具 (Function Calling)
"""
print(11111)
messages = [{"role": "user", "content": query}]
print(2222)
response = await self.session.list_tools()
print(33333)
available_tools = [{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"input_schema": tool.inputSchema
}
} for tool in response.tools]
print(available_tools)
print(44)
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=available_tools
)
print(5555)
# 處理返回的內容
content = response.choices[0]
if content.finish_reason == "tool_calls":
# 如何是需要使用工具，就解析工具
tool_call = content.message.tool_calls[0]
tool_name = tool_call.function.name
tool_args = json.loads(tool_call.function.arguments)
# 執行工具
result = await self.session.call_tool(tool_name, tool_args)
print(f"nn[Calling tool {tool_name} with args {tool_args}]nn")
# 將模型返回的調用哪個工具數據和工具執行完成后的數據都存入messages中
messages.append(content.message.model_dump())
messages.append({
"role": "tool",
"content": result.content[0].text,
"tool_call_id": tool_call.id,
})
# 將上面的結果再返回給大模型用于生產最終的結果
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
)
return response.choices[0].message.content
return content.message.content
# async def process_query(self, query: str) -> str:
# """調用 OpenAI API 處理用戶查詢"""
# messages = [{"role": "system", "content": "你是一個智能助手，幫助用戶回答問題。"},
# {"role": "user", "content": query}]
# try:
# # 調用 OpenAI API
# response = await asyncio.get_event_loop().run_in_executor(
# None,
# lambda: self.client.chat.completions.create(
# model=self.model,
# messages=messages
# )
# )
# return response.choices[0].message.content
# except Exception as e:
# return f"?? 調用 OpenAI API 時出錯: {str(e)}"
async def chat_loop(self):
"""運行交互式聊天循環"""
print("n?? MCP 客戶端已啟動！輸入 'quit' 退出")
while True:
try:
query = input("n你: ").strip()
if query.lower() == 'quit':
break
print("chat ing")
response = await self.process_query(query)
# 發送用戶輸入到 OpenAI API
print(f"n?? OpenAI: {response}")
except Exception as e:
print(f"n?? 發生錯誤: {str(e)}")
async def cleanup(self):
"""清理資源"""
await self.exit_stack.aclose()
async def main():
if len(sys.argv) < 2:
print("Usage: python client.py ")
sys.exit(1)
client = MCPClient()
try:
await client.connect_to_server(sys.argv[1])
await client.chat_loop()
finally:
await client.cleanup()
if __name__ == "__main__":
import sys
asyncio.run(main())要么新啟動一個終端來運行server部分,需要重新進入虛擬環境.當然,uv也支持同時運行,所以可以直接運行:
uv run client_new.py server.py
最終效果:
文章轉載自： MCP客戶端與服務端初使用——讓deepseek調用查詢天氣的mcp來查詢天氣