【原】客户端创建实战 - Claude MCP 大模型上下文协议入门实战

小张学AI 2025-03-11 发布于山东

展开全文

· 大家好，我是 同学小张，日常分享AI知识和实战案例
· 欢迎 点赞 + 关注 👏，持续学习，持续干货输出。
· +v: jasper_8017 一起交流💬，一起进步💪，更有专业资料领取！
微信私信免费领取AI、C++等相关资料，持续收集更新中！包括但不限于：

上篇文章，我们从零到一利用MCP协议，创建了一个MCP服务端，并利用 Claude 客户端连接了服务端完成了服务中Tool的调用。今天，这篇文章，我们继续学习MCP协议，利用MCP协议从零到一写一个客户端，并将客户端与服务端连接，调用服务端的Tool。

1. MCP环境配置

2.1 依赖

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

2.2 创建环境

（1）安装uv

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

（2）利用uv创建并设置项目

# 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 配置大模型

因为客户端需要用到大模型，所以需要设置 API Key。像往常一样，我们还是将API Key放到本地的.env文件中。

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

官方示例中使用的是 ANTHROPIC_API_KEY。

ANTHROPIC_API_KEY=<your key here>

我没有... 所以咱们还是用原来的 OpenAI API key进行代替。

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

2. 实战

2.1 创建 MCP 客户端

使用 mcp 中的 ClientSession，构建一个客户端session。

from mcp import ClientSession, StdioServerParameters

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

2.2 连接 MCP 服务器

2.2.1 代码编写

connect_to_server 函数用来实现客户端与服务端的连接。server_script_path 为 MCP 服务的脚本地址，也就是上篇文章中的 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）服务的连接方式：

首先定义一个 StdioServerParameters 对象：

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

然后利用 self.exit_stack.enter_async_context(stdio_client(server_params)) 来运行。这种方式是用 stdio 方式运行，就是在当前客户端运行进程中，开一个子进程来运行服务端。大家可以先简单了解一下，先跑通流程为主。

（2）客户端的运行：

先像服务端一样利用 enter_async_context 运行，获取 sesssion，然后对 session 进行初始化。

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

await self.session.initialize()

（3）self.session.list_tools() 用来获取连接的服务端有哪些可用的工具。获取到的结果如下：注意这里返回的tools的格式，与OpenAI的function calling的格式不一样。

2.3 客户端处理用户问题

以下代码用来处理用户输入。

（1）接收用户输入

（2）调用大模型，让大模型决定是否调用工具，调用工具的名称和参数。这里注意：要将 tools参数填上，才能利用大模型的function calling能力获取到应该调用哪个工具。

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

还有，注意，这里因为我们用的是OpenAI的大模型，其Function calling传递tool的格式与MCP服务返回的tool list的格式不一致，所以要将其进行转换，封装成OpenAI大模型认可的tools格式。

具体格式可以看我这篇文章：【AI大模型应用开发】2.1 Function Calling连接外部世界 - 入门与实战（1）
https://blog.csdn.net/Attitude93/article/details/135906466?spm=1011.2415.3001.5331

我这里简单转换一下（没转换完，剩下的参数部分大家可以试着自己转换）：

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

（3）如果大模型决定需要Tool，则调用 MCP 服务上的 Tool：

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

最终，完整的 process_query 函数如下：

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"调用 {tool_call.function.name} 函数，参数是 {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 连续对话封装

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函数封装

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. 运行

在终端命令行中运行，运行命令如下：

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

运行成功后，输出如下，可以体验了。

4. 总结

4.1 关键组件

· MCPClient 类在初始化时会进行会话管理和 API 客户端的配置。
· 使用 AsyncExitStack 来妥善管理资源。
· 配置 OpenAI API Key 以实现与 GPT模型的交互。

4.2 总结

本文我们从零到一实现了一个MCP客户端，并与我们上一篇文章中从零到一实现的MCP服务端进行了连接和通信。这期间，除了上面提到的关键组件外，额外需要注意的是，如果你不是用的 Claude 模型，需要格外注意你所使用的大模型对function calling能力的支持，以及其需要的function calling的格式。否则很可能模型无法识别到需要用哪个tool，或者在调用tool时出错。