本文详细介绍了如何构建一个基于 MCP (Message Communication Protocol) 的 Studio，涵盖了环境准备、代码开发、以及客户端和服务端的交互。通过注册智谱 AI 账号、编写服务端和客户端代码，并使用 MCP 框架进行工具的定义和调用，最终实现了一个简单的问候语生成和加法计算的功能。

🔑 准备工作：首先，需要注册智谱 AI 并获取 API 密钥，以便后续调用其模型。

⚙️ 服务端代码：服务端代码使用 FastMCP 框架，定义了工具函数 sayHello 用于问候用户，以及 get_data_add 用于执行加法运算。此外，还定义了资源函数 list_test_dir 用于列出目录中的文件。

💻 客户端代码：客户端代码通过 MCPClient 类与服务端建立连接，并使用智谱 AI 模型来调用工具。客户端首先列出可用的工具，然后根据用户提出的问题，调用相应的工具并处理返回结果，最终得到最终的回复。

💡 核心流程：客户端连接到服务器后，通过调用 list_tools 获取可用工具列表。用户输入问题后，客户端将问题和可用工具传递给智谱 AI 模型。模型根据问题选择合适的工具，客户端调用工具并获取结果，最终将结果返回给用户。

你好，我是 shengjk1，多年大厂经验，努力构建 通俗易懂的、好玩的编程语言教程。 欢迎关注！你会有如下收益：

了解大厂经验拥有和大厂相匹配的技术等

希望看什么，评论或者私信告诉我！

一、背景

前面我们已经再为更进一步的 MCP 打下了基础一文搞定 Python 装饰器

解锁 MCP 中的 JSON-RPC

为什么MCP可以适配不同LLM

MCP协议三种传输机制全解析

多智能体大语言模型系统频频翻车？三大失败根源与解决方案全解析

今天就正式进入到代码开发阶段

二、准备工作

2.1 注册 zhipu

通过专属链接获取 2000万Tokens

2.2 进入控制台，生成秘钥

2.3 zhipu 例子

pip install zhipuai

# -*- coding: utf-8 -*-#%%from zhipuai import ZhipuAIclient = ZhipuAI(api_key="xxxxxxx")  # 请填写您自己的APIKeyresponse = client.chat.completions.create(    model="glm-4-flash-250414",  # 请填写您要调用的模型名称    messages=[        {"role": "user", "content": "一个袋子中有5个红球和3个蓝球,随机抽取2个球,抽到至少1个红球的概率为:"}    ],    max_tokens=12000,)# print(response.choices[0]['message']['content'])print(response)print(response.choices[0].message.content)

三、MCP Stido 代码

3.1 mcp server 代码

# -*- coding: utf-8 -*-# create by shengjk1 on  2025/6/5import pathlibfrom typing import Anyfrom mcp.server import FastMCP# Initialize FastMCP servermcp = FastMCP("test")@mcp.tool()async  def sayHello(name:str)->str:    """     greet a user by name.    :param name: The name of the user to greet.    :return: A greeting message.    """    return f"Hello, {name}!"@mcp.tool()async def get_data_add(a: int,b:int) -> Any:    """    Fetch data from a given URL.    :param url: The URL to fetch data from. Default is 'https://api.github.com'.    :return: The JSON data fetched from the URL.    """    return a + b@mcp.resource("dir://mcp_demo-quickstart")def list_test_dir() -> list[str]:    """List the files in the test directory"""    # Iterate through the test directory and return a list of strings representing the files    return [str(f) for f in pathlib.Path('mcp_demo-quickstart').iterdir()]if __name__ == '__main__':    mcp.run(transport='stdio')

3.2 mcp server 代码详细解释

3.2.1. 导入模块

import pathlibfrom typing import Anyfrom mcp.server import FastMCP

pathlib 是 Python 的标准库模块，用于处理文件路径和目录操作。typing.Any 是一个类型注解，表示函数返回值可以是任意类型。mcp.server.FastMCP 是从 mcp 框架中导入的 FastMCP 类，用于创建和管理服务器。

3.2.2. 初始化 FastMCP 服务器

mcp = FastMCP("test")

创建了一个 FastMCP 服务器实例，命名为 "test"。这个实例将用于注册工具函数和资源，并启动服务器。

3.2.3 定义工具函数

使用 @mcp.tool() 装饰器将 sayHello 函数注册为一个工具函数。( 扩展知识：一文搞定 Python 装饰器 )需要注意使用 @mcp.tool() 装饰器装饰的函数，一定写好函数的 doc: 描述清楚函数的作用和参数 这个很重要，不然 tool call 的时候，没有办法选择合适的工具

3.2.4. 定义资源函数

使用 @mcp.resource() 装饰器将 list_test_dir 函数注册为一个资源函数。同 @mcp.tool() 要描述清楚 doc

3.2.5. 启动服务器

if __name__ == '__main__':    mcp.run(transport='stdio')

当脚本作为主程序运行时，调用 mcp.run() 方法启动服务器。参数 transport='stdio' 表示服务器通过标准输入输出（stdio）进行通信。

3.2.6. 总结

这段代码实现了一个简单的 mcp 服务器，提供了以下功能：

工具函数：
sayHello：根据用户的名字生成问候语。get_data_add：计算两个数字的和（但注释中提到的功能与实际实现不符）。
资源函数：
list_test_dir：列出指定目录中的所有文件。
服务器启动：
使用 stdio 作为通信方式。

3.3 mcp client 代码

# -*- coding: utf-8 -*-import asyncioimport jsonfrom typing import Optionalfrom contextlib import AsyncExitStackfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientfrom mcp.types import TextContent, ImageContent, EmbeddedResourceclass MCPClient:    def __init__(self):        self.session:Optional[ClientSession] = None        self.exit_stack=AsyncExitStack()    async def connect_to_server(self,server_script_path: str):        command = "python"        server_params = StdioServerParameters(            command=command,            args=[server_script_path],            env=None        )        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()        response=await self.session.list_tools()        tools = response.tools        print("===========",tools)    async def close(self):        await self.exit_stack.aclose()    async def process_query(self, query: str) -> str:        """Process a query using available tools"""        messages = [            {                "role": "user",                "content": query            }        ]        available_tools = [{            "type": "function",            "function":{                "name": tool.name,                "description": tool.description,                "parameters": tool.inputSchema            }        } for tool in response.tools]        print("Available tools:", available_tools)        from zhipuai import ZhipuAI        zhipu_client = ZhipuAI(api_key="xxxxxx")  # 请填写您自己的APIKey        response = zhipu_client.chat.completions.create(            model="glm-4-flash-250414",  # 请填写您要调用的模型名称            messages=messages,            max_tokens=12000,            tools=available_tools        )        print("response:",response)        if response.choices[0].finish_reason == "tool_calls":            tool_calls = response.choices[0].message.tool_calls            print("Tool calls:", tool_calls)            for tool_call in tool_calls:                tool_name = tool_call.function.name                tool_input = json.loads(tool_call.function.arguments)                print(f"Calling tool {tool_name} with input {tool_input}")                # 在这里调用相应的工具，并获取结果                result = await self.session.call_tool(tool_name, tool_input)                #Tool get_data_add returned result: meta=None content=[TextContent(type='text', text='11', annotations=None)] isError=False                print(f"Tool {tool_name} returned result: {result.content}")                messages.append({                    "role": "user",                    "content": ','.join(contens.text for contens in result.content)                })                for item in result.content:                    if isinstance(item, TextContent):                        # 将工具调用的结果添加到消息列表中                        messages.append({                            "role": "assistant",                            "content": item.text                        })            print("Updated messages:", messages)            response = zhipu_client.chat.completions.create(                model="glm-4-flash-250414",  # 请填写您要调用的模型名称                messages=messages,                max_tokens=12000,            )            print("response:",response.choices[0].message.content)async def main():    client = MCPClient()    try:        await client.connect_to_server("../server/server.py")        await client.process_query("两步走，首先像小明问好，其次并把1和10相加，最后返回问好和加法的结果")    finally:        await client.close()if __name__ == '__main__':    asyncio.run(main())

3.4 mcp client 代码解释

这段代码是一个简化版本的 MCPClient 类实现，它结合了与服务器的通信（通过 mcp 模块）和智谱AI（ZhipuAI）的接口调用。以下是对代码的详细解释：

3.4.1. MCPClient 类的定义

class MCPClient:    def __init__(self):        self.session: Optional[ClientSession] = None        self.exit_stack = AsyncExitStack()

self.session：用于存储与服务器的会话对象，初始值为 None。self.exit_stack：用于管理异步上下文，确保在程序结束时正确关闭资源。

3.4.2. 连接到服务器

async def connect_to_server(self, server_script_path: str):    command = "python"    server_params = StdioServerParameters(        command=command,        args=[server_script_path],        env=None    )    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()    response = await self.session.list_tools()    tools = response.tools    print("===========", tools)

server_script_path：服务器脚本的路径。command：用于启动服务器的命令（这里是 python）。server_params：定义了服务器的启动参数，包括命令、参数列表和环境变量。stdio_transport：通过 stdio_client 创建一个标准输入输出的传输通道。self.stdio, self.write：从 stdio_transport 中获取输入和输出流。self.session：创建一个 ClientSession 对象，用于与服务器进行通信。await self.session.initialize()：初始化会话。await self.session.list_tools()：获取服务器支持的工具列表并打印。

3.4.3. 关闭客户端

async def close(self):    await self.exit_stack.aclose()

await self.exit_stack.aclose()：关闭异步上下文管理器，确保所有资源被正确释放。

3.4.4. 处理查询

async def process_query(self, query: str) -> str:    messages = [        {            "role": "user",            "content": query        }    ]

query：用户输入的查询字符串。messages：初始化消息列表，包含用户的消息。

构建可用工具列表

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

将工具信息转换为特定格式，用于后续的工具调用。

调用智谱AI接口

from zhipuai import ZhipuAIzhipu_client = ZhipuAI(api_key="xxxxxx")  # 替换为实际的API Keyresponse = zhipu_client.chat.completions.create(    model="glm-4-flash-250414",  # 模型名称    messages=messages,    max_tokens=12000,    tools=available_tools)print("response:", response)

使用智谱AI的接口，根据用户输入和可用工具生成响应。

处理工具调用

if response.choices[0].finish_reason == "tool_calls":    tool_calls = response.choices[0].message.tool_calls    print("Tool calls:", tool_calls)    for tool_call in tool_calls:        tool_name = tool_call.function.name        tool_input = json.loads(tool_call.function.arguments)        print(f"Calling tool {tool_name} with input {tool_input}")        result = await self.session.call_tool(tool_name, tool_input)        print(f"Tool {tool_name} returned result: {result.content}")        messages.append({            "role": "user",            "content": ','.join(contens.text for contens in result.content)        })        for item in result.content:            if isinstance(item, TextContent):                messages.append({                    "role": "assistant",                    "content": item.text                })

如果响应中包含工具调用请求：
遍历工具调用列表，调用相应的工具并获取结果。将工具调用的结果添加到消息列表中。根据结果类型（文本、图片、嵌入资源）进行处理。

重新调用智谱AI接口

response = zhipu_client.chat.completions.create(    model="glm-4-flash-250414",    messages=messages,    max_tokens=12000,)print("response:", response.choices[0].message.content)

使用更新后的消息列表重新调用智谱AI接口，生成最终的响应。

3.4.5. 主函数

async def main():    client = MCPClient()    try:        await client.connect_to_server("../server/server.py")        await client.process_query("两步走，首先像小明问好，其次并把1和10相加，最后返回问好和加法的结果")    finally:        await client.close()if __name__ == '__main__':    asyncio.run(main())

创建 MCPClient 实例。连接到服务器并处理用户查询。确保在程序结束时关闭客户端。

代码逻辑总结

连接到服务器：通过 connect_to_server 方法启动服务器脚本，并建立与服务器的通信会话。处理用户查询：
将用户输入的查询包装为消息格式。调用智谱AI接口，获取初步响应。如果响应中包含工具调用请求，则调用服务器提供的工具。将工具调用的结果更新到消息列表中。重新调用智谱AI接口，生成最终的响应。
关闭客户端：确保在程序结束时释放所有资源。

注意事项

服务器脚本路径："../server/server.py" 是服务器脚本的路径，需要根据实际情况调整。工具调用：工具调用的结果需要根据实际返回的 result.content 进行处理。

四、执行和结果

执行运行 client.py 文件即可，client 会通过下面的代码以子进程的方式启动 server

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

结果

response: 你好，小明！1和10相加等于11。

五、总结

文章详细讲解了如何使用 MCP 协议和智谱 AI 平台构建一个工具调用系统。通过注册智谱 AI 平台获取 API Key，并编写了 MCP Server 和 Client 的代码。Server 提供了工具函数，Client 通过智谱 AI 的接口调用这些工具函数，并处理用户查询。代码逻辑清晰，包含了工具函数的定义、服务器的启动、客户端的连接以及查询处理等关键步骤。