這個的思路很簡單就不多介紹了，主要是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/activate

windows中則是:

myenv\Scripts\activate     # Windows

然后在虛擬環境中都要通過uv開頭的命令來運行文件,這樣能夠更加高效.

上面主要是uv的一些基礎內容,接下來我們正式建立mcp client和mcp server
創建mcp客戶端:
我用的是自己在win上安裝的py3.8

uv init mcp-client

cd mcp-client

然后創建虛擬環境并進入:

uv venv

.venv\Scripts\activate

接下來安裝一些依賴庫:

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

.venv\Scripts\activate

uv add mcp openai python-dotenv

python雙重環境…
安裝完畢之后:

創建客戶端client.py,是調用大模型的核心部分:

import asyncio

import os

from openai import OpenAI

from dotenv import load_dotenv

from contextlib import AsyncExitStack



# 加載 .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) 



    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,

                    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:

import json

import httpx

from typing import Any

from mcp.server.fastmcp import FastMCP



# 初始化 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": 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}°C\n"

        f"?? 濕度: {humidity}%\n"

        f"?? 風速: {wind_speed} m/s\n"

        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:

import asyncio

import os

import json

from typing import Optional

from contextlib import AsyncExitStack



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"\n\n[Calling tool {tool_name} with args {tool_args}]\n\n")



            # 將模型返回的調用哪個工具數據和工具執行完成后的數據都存入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 <path_to_server_script>")

        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來查詢天氣

上一篇：

Python與Ollama的開發案例

下一篇：

AI零門檻變現秘籍：最新賺錢公式全公開
#你可能也喜歡這些API文章!

我們有何不同？

API服務商零注冊

多API并行試用

數據驅動選型，提升決策效率

查看全部API→
??

熱門場景實測，選對API

#AI文本生成大模型API

對比大模型API的內容創意新穎性、情感共鳴力、商業轉化潛力

25個渠道
一鍵對比試用API 限時免費

#AI深度推理大模型API

對比大模型API的邏輯推理準確性、分析深度、可視化建議合理性

10個渠道
一鍵對比試用API 限時免費