• · 大家好，我是 同學(xué)小張，日常分享AI知識(shí)和實(shí)戰(zhàn)案例

• · 歡迎 點(diǎn)贊 + 關(guān)注 ??，持續(xù)學(xué)習(xí)，持續(xù)干貨輸出。

• · +v: jasper_8017 一起交流??，一起進(jìn)步??，更有專(zhuān)業(yè)資料領(lǐng)??！

• 微信私信免費(fèi)領(lǐng)取AI、C++等相關(guān)資料，持續(xù)收集更新中！包括但不限于：

上篇文章，我們從零到一利用MCP協(xié)議，創(chuàng)建了一個(gè)MCP服務(wù)端，并利用 Claude 客戶(hù)端連接了服務(wù)端完成了服務(wù)中Tool的調(diào)用。今天，這篇文章，我們繼續(xù)學(xué)習(xí)MCP協(xié)議，利用MCP協(xié)議從零到一寫(xiě)一個(gè)客戶(hù)端，并將客戶(hù)端與服務(wù)端連接，調(diào)用服務(wù)端的Tool。

1. MCP環(huán)境配置

2.1 依賴(lài)

Python 3.10或更高版本
Python MCP SDK 1.2.0或更高版本

2.2 創(chuàng)建環(huán)境

（1）安裝uv

curl -LsSf https:///uv/install.sh | sh

（2）利用uv創(chuàng)建并設(shè)置項(xiàng)目

# Create project directory
uv init mcp-client
cd mcp-client

# Create virtual environment
uv venv

# Activate virtual environment
# On Windows:
.venv\Scripts\activate
# On Unix or MacOS:
source .venv/bin/activate

# Install required packages
uv add mcp anthropic python-dotenv

# Remove boilerplate files
rm hello.py

# Create our main file
touch client.py

2.3 配置大模型

因?yàn)榭蛻?hù)端需要用到大模型，所以需要設(shè)置 API Key。像往常一樣，我們還是將API Key放到本地的.env文件中。

# Create .env file
touch .env
echo ".env" >> .gitignore

官方示例中使用的是 ANTHROPIC_API_KEY。

ANTHROPIC_API_KEY=<your key here>

我沒(méi)有... 所以咱們還是用原來(lái)的 OpenAI API key進(jìn)行代替。

OPENAI_API_KEY = "sk-xxxxxx"
OPENAI_BASE_URL = "https://api./v1"

2. 實(shí)戰(zhàn)

2.1 創(chuàng)建 MCP 客戶(hù)端

使用 mcp 中的 ClientSession，構(gòu)建一個(gè)客戶(hù)端session。

from mcp import ClientSession, StdioServerParameters

class MCPClient:
    def __init__(self):
        # Initialize session and client objects
        self.session: Optional[ClientSession] = None

2.2 連接 MCP 服務(wù)器

2.2.1 代碼編寫(xiě)

connect_to_server 函數(shù)用來(lái)實(shí)現(xiàn)客戶(hù)端與服務(wù)端的連接。server_script_path 為 MCP 服務(wù)的腳本地址，也就是上篇文章中的 weather.py的路徑。

async defconnect_to_server(self, server_script_path: str):
    """Connect to an MCP server

    Args:
        server_script_path: Path to the server script (.py or .js)
    """
    is_python = server_script_path.endswith('.py')
    is_js = server_script_path.endswith('.js')
    ifnot (is_python or is_js):
        raise ValueError("Server script must be a .py or .js file")

    command = "python"if is_python else"node"
    server_params = StdioServerParameters(
        command=command,
        args=[server_script_path],
        env=None
    )

    stdio_transport = awaitself.exit_stack.enter_async_context(stdio_client(server_params))
    self.stdio, self.write = stdio_transport
    self.session = awaitself.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))

    awaitself.session.initialize()

    # List available tools
    response = awaitself.session.list_tools()
    tools = response.tools
    print("\nConnected to server with tools:", [tool.name for tool in tools])

2.2.2 代碼解釋

（1）服務(wù)的連接方式：

首先定義一個(gè) StdioServerParameters 對(duì)象：

server_params = StdioServerParameters(
    command=command,
    args=[server_script_path],
    env=None
)

然后利用 self.exit_stack.enter_async_context(stdio_client(server_params)) 來(lái)運(yùn)行。這種方式是用 stdio 方式運(yùn)行，就是在當(dāng)前客戶(hù)端運(yùn)行進(jìn)程中，開(kāi)一個(gè)子進(jìn)程來(lái)運(yùn)行服務(wù)端。大家可以先簡(jiǎn)單了解一下，先跑通流程為主。

（2）客戶(hù)端的運(yùn)行：

先像服務(wù)端一樣利用 enter_async_context 運(yùn)行，獲取 sesssion，然后對(duì) session 進(jìn)行初始化。

self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))

await self.session.initialize()

（3）self.session.list_tools() 用來(lái)獲取連接的服務(wù)端有哪些可用的工具。獲取到的結(jié)果如下：注意這里返回的tools的格式，與OpenAI的function calling的格式不一樣。

2.3 客戶(hù)端處理用戶(hù)問(wèn)題

以下代碼用來(lái)處理用戶(hù)輸入。

（1）接收用戶(hù)輸入

（2）調(diào)用大模型，讓大模型決定是否調(diào)用工具，調(diào)用工具的名稱(chēng)和參數(shù)。這里注意：要將 tools參數(shù)填上，才能利用大模型的function calling能力獲取到應(yīng)該調(diào)用哪個(gè)工具。

response = self.openai.chat.completions.create(
    model="gpt-4o",
    max_tokens=1000,
    messages=messages,
    tools=available_tools
)

還有，注意，這里因?yàn)槲覀冇玫氖荗penAI的大模型，其Function calling傳遞tool的格式與MCP服務(wù)返回的tool list的格式不一致，所以要將其進(jìn)行轉(zhuǎn)換，封裝成OpenAI大模型認(rèn)可的tools格式。

具體格式可以看我這篇文章：【AI大模型應(yīng)用開(kāi)發(fā)】2.1 Function Calling連接外部世界 - 入門(mén)與實(shí)戰(zhàn)（1）

https://blog.csdn.net/Attitude93/article/details/135906466?spm=1011.2415.3001.5331

我這里簡(jiǎn)單轉(zhuǎn)換一下（沒(méi)轉(zhuǎn)換完，剩下的參數(shù)部分大家可以試著自己轉(zhuǎn)換）：

available_tools = [{
    "type": "function",
    "function": {
        "name": tool.name,
        "description": tool.description,
        "input_schema": tool.inputSchema
    }
} for tool in response.tools]

（3）如果大模型決定需要Tool，則調(diào)用 MCP 服務(wù)上的 Tool：

result = await self.session.call_tool(tool_name, tool_args)

最終，完整的 process_query 函數(shù)如下：

async defprocess_query(self, query: str) -> str:
        """Process a query using OpenAI and available tools"""
        messages = [
            {
                "role": "user",
                "content": query
            }
        ]
        
        response = awaitself.session.list_tools()
        available_tools = [{
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "input_schema": tool.inputSchema
            }
        } for tool in response.tools]
        
        print("Available Tools:" + str(available_tools))
        response = self.openai.chat.completions.create(
            model="gpt-4o",
            max_tokens=1000,
            messages=messages,
            tools=available_tools
        )
        
        response = response.choices[0].message

        # Process response and handle tool calls
        tool_results = []
        final_text = []

        assistant_message_content = []
        print(response)
        if (response.tool_calls isnotNone):
            for tool_call in response.tool_calls:
                print(response.tool_calls)
                print(f"調(diào)用 {tool_call.function.name} 函數(shù)，參數(shù)是 {tool_call.function.arguments}")
                tool_name = tool_call.function.name
                tool_args = tool_call.function.arguments

                # Execute tool call
                result = awaitself.session.call_tool(tool_name, tool_args)
                tool_results.append({"call": tool_name, "result": result})
                final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")

                assistant_message_content.append(content)
                messages.append({
                    "role": "assistant",
                    "content": assistant_message_content
                })
                messages.append({
                    "role": "user",
                    "content": [
                        {
                            "type": "tool_result",
                            "tool_use_id": content.id,
                            "content": result.content
                        }
                    ]
                })

                # Get next response from OpenAI
                response = self.openai.chat.completions.create(
                    model="gpt-3.5-turbo",
                    max_tokens=1000,
                    messages=messages,
                    tools=available_tools
                )
                print("response:\n")
                print(response)
                final_text.append(response.choices[0].message.content)
        
        
        for content in response.choices[0].message.content:
            final_text.append(content)
            assistant_message_content.append(content)

        return"\n".join(final_text)

2.4 連續(xù)對(duì)話(huà)封裝

async defchat_loop(self):
    """Run an interactive chat loop"""
    print("\nMCP Client Started!")
    print("Type your queries or 'quit' to exit.")

    whileTrue:
        try:
            query = input("\nQuery: ").strip()

            if query.lower() == 'quit':
                break

            response = awaitself.process_query(query)
            print("\n" + response)

        except Exception as e:
            print(f"\nError: {str(e)}")

asyncdefcleanup(self):
    """Clean up resources"""
    awaitself.exit_stack.aclose()

2.5 main函數(shù)封裝

async defmain():
    iflen(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())

3. 運(yùn)行

在終端命令行中運(yùn)行，運(yùn)行命令如下：

uv run client.py xxxxxx/weather/weather.py

4. 總結(jié)

4.1 關(guān)鍵組件

• · MCPClient 類(lèi)在初始化時(shí)會(huì)進(jìn)行會(huì)話(huà)管理和 API 客戶(hù)端的配置。
• · 使用 AsyncExitStack 來(lái)妥善管理資源。
• · 配置 OpenAI API Key 以實(shí)現(xiàn)與 GPT模型 的交互。

4.2 總結(jié)

本文我們從零到一實(shí)現(xiàn)了一個(gè)MCP客戶(hù)端，并與我們上一篇文章中從零到一實(shí)現(xiàn)的MCP服務(wù)端進(jìn)行了連接和通信。這期間，除了上面提到的關(guān)鍵組件外，額外需要注意的是，如果你不是用的 Claude 模型，需要格外注意你所使用的大模型對(duì)function calling能力的支持，以及其需要的function calling的格式。否則很可能模型無(wú)法識(shí)別到需要用哪個(gè)tool，或者在調(diào)用tool時(shí)出錯(cuò)。

5. 參考

• · https:///quickstart/client
• · 完整代碼：https://github.com/modelcontextprotocol/quickstart-resources/tree/main/mcp-client

如果覺(jué)得本文對(duì)你有幫助，麻煩點(diǎn)個(gè)贊和關(guān)注唄 ~~~

私信免費(fèi)領(lǐng)取AI、C++等相關(guān)資料，持續(xù)收集更新中！ 包括但不限于：