本文详细介绍如何使用Python开发MCP Client，支持stdio、sse、streamable-http所有传输类型。提供完整实现代码，包含异步与非异步方法，展示工具定义格式转换及实际调用示例。为开发者提供大模型Agent工具调用全面解决方案，适合有一定Python基础的学习者。

背景

MCP（Model Context Protocol，模型上下文协议）的出现，统一了Agent工具使用的方式，虽然市面上绝大多数MCP存在问题，但不可否认依然还是有非常多优秀好用的MCP工具，关于MCP协议更多的内容，后续会有专门的文章介绍，本文先聚焦在如何使用MCP工具上。

本文会涉及如何使用Python开发MCP Client，并使用LLM完成一次工具调用示例，支持MCP目前支持的所有传输类型，包括stdio、sse、streamable-http，也就是说，市面上所有类型的MCP Server，都是支持的。

总的调用方式，其实与我们之前介绍的常规工具大同小异，只是需要一些准备工作，总体流程如下：

准备工作

本文准备工作与动手学Agent：工具使用(2)工具使用基础&强制模型选择特定工具中介绍的基本一致，包括：

  • API申请：申请高德API Key、秘塔搜API，这两个都不是必须的，是为了演示sse类型和streamable-http类型的传输类型需要的，可以根据需要申请
  • 模型准备：LM Studio中下载qwen3 8b，也可以使用Ollama或者在线服务
  • 安装依赖

核心代码

本文代码已开源，地址在：https://github.com/Steven-Luo/hands-on-agents。

MCP官方库给出的样例代码，都是异步函数，为了兼顾到不同的使用场景，本文进行了扩展，参考langchain的命名方式，alist_tools、acall_tool都是异步方法，需要使用await alist_tools(...)这样的方式进行调用，如果你的调用方是非异步函数，则可以使用list_tools、call_tools。

由于async方法是事件循环驱动的，Jupyter Notebook启动后，本身就已经有一个事件循环在运行了，asyncio默认不允许事件循环出现嵌套，因此在Jupyter Notebook中调用async方法，需要在调用前执行如下代码：

import nest_asyncio

nest_asyncio.apply()

MCP Client

完整的MCP Client实现如下：

from contextlib import asynccontextmanager
from typing import List

from mcp import ClientSession, Tool
from mcp.client.streamable_http import streamablehttp_client
from mcp.client.sse import sse_client
from mcp.client.stdio import stdio_client, StdioServerParameters
import asyncio

classMCPClient:
def__init__(self, mcp_servers: dict, timeout: int=60):
"""
        Args:
            mcp_servers: MCP Server配置字典，格式参考Claude MCP Server定义格式(https://docs.claude.com/en/docs/claude-code/mcp?utm_source=chatgpt.com)，格式如下：
            {
                "mcpServers": {
                    "mcp-server-chart": {
                        "command": "npx",
                        "args": [
                            "-y",
                            "@antv/mcp-server-chart"
                        ]
                    },
                    "mySseServer": {
                        "url": "https://api.example.com/sse",
                        "type": "sse",
                        "headers": {
                            "Authorization": "Bearer your_api_key"
                        }
                    },
                    "myHttpServer": {
                        "type": "http",
                        "url": "https://api.example.com/mcp",
                        "headers": {
                            "Authorization": "Bearer your_api_key"
                        }
                    }
                }
            }
        """

        self.mcp_servers = mcp_servers['mcpServers']
        self.timeout = timeout

def_get_mcp_client(self, mcp_server_name):
        server_config = self.mcp_servers[mcp_server_name]
if'url'in server_config:
            mcp_timeout = float(self.timeout)
            logger.info(f"mcp_timeout: {mcp_timeout}")

            url = server_config['url']
            headers = server_config.get('headers', {})
if server_config['type'] == 'sse':
                client = sse_client(url=url, headers=headers, timeout=mcp_timeout)
else:
                client = streamablehttp_client(url=url, headers=headers, timeout=mcp_timeout)
else:
            server_params = StdioServerParameters(
                command=server_config['command'],
                args=server_config['args'],
                env=server_config.get('env')
            )
            client = stdio_client(server_params)
return client

    @asynccontextmanager
asyncdef_get_transport(self, mcp_server_name):
        server_config = self.mcp_servers[mcp_server_name]
# logger.info(f"mcp_server_name: {mcp_server_name}, server_config: {server_config}")

if'url'notin server_config and'command'notin server_config:
raise ValueError(f"Invalid MCP server config: {server_config}")

if'url'in server_config:
            mcp_timeout = float(self.timeout)
            logger.info(f"mcp_timeout: {mcp_timeout}")

            url = server_config['url']
            headers = server_config.get('headers', {})
if server_config['type'] == 'sse':
                client = sse_client(url=url, headers=headers, timeout=mcp_timeout)
else:
                client = streamablehttp_client(url=url, headers=headers, timeout=mcp_timeout)
else:
            server_params = StdioServerParameters(
                command=server_config['command'],
                args=server_config['args'],
                env=server_config.get('env')
            )
            client = stdio_client(server_params)

asyncwith client as session_input:
# streamablehttp_client返回的session_input包含get_session_id函数
if server_config['type'] == 'http':
                read_stream, write_stream, get_session_id = session_input
else:
                read_stream, write_stream = session_input
yield read_stream, write_stream

asyncdefalist_tools(self, mcp_server_name):
"""
        列出指定MCP Server上可用的工具(异步)
        Returns:
            可用的工具列表
        """
        logger.info(f"mcp_server_name: {mcp_server_name}")
        tools = []

asyncwith self._get_transport(mcp_server_name) as (read_stream, write_stream):
asyncwith ClientSession(read_stream, write_stream) as session:
await session.initialize()
                response = await session.list_tools()
                tools = response.tools
return tools

asyncdefacall_tool(self, mcp_server_name, tool_name: str, tool_args: dict={}):
"""
        调用指定MCP Server上的工具(异步)
        Args:
            tool_name: 工具名称
            tool_args: 工具参数
        Returns:
            工具调用结果
        """
        logger.info(f"call_tool, server_name: {mcp_server_name}, tool_name: {tool_name}, tool_args: {tool_args}")

asyncwith self._get_transport(mcp_server_name) as (read_stream, write_stream):
# 使用客户端流创建会话
asyncwith ClientSession(read_stream, write_stream) as session:
# 初始化连接
await session.initialize()
                result = await session.call_tool(tool_name, tool_args)
return result

deflist_tools(self, mcp_server_name):
return asyncio.run(self.alist_tools(mcp_server_name))

defcall_tool(self, mcp_server_name, tool_name: str, tool_args: dict={}):
return asyncio.run(self.acall_tool(mcp_server_name, tool_name, tool_args))

采用与Claude相同的配置格式，我们使用三种类型的MCP Server进行实验：

  • mcp-server-chart：这是蚂蚁集团开发的可视化MCP Server，stdio类型
  • 高德地图MCP Server：sse类型
  • 秘塔搜MCP Server：streamable-http类型

获取工具

config_data = {
"mcpServers": {
"mcp-server-chart": {
"command": "npx",
"args": [
"-y",
"@antv/mcp-server-chart"
            ]
        },
"amap": {
"url": f"https://mcp.amap.com/sse?key={os.getenv('AMAP_API_KEY')}",
"type": "sse"
        },
"metaso": {
"type": "http",
"url": "https://metaso.cn/api/mcp",
"headers": {
"Authorization": f'Bearer {os.getenv("METASO_API_KEY")}'
            }
        }
    }
}
mcp_client = MCPClient(config_data)
# 使用异步方法获取工具
tools = await mcp_client.alist_tools('metaso')
# 使用非异步方法获取工具
tools = mcp_client.list_tools('metaso')
# 查看工具定义的格式
[item.model_dump() for item in tools]

MCP工具定义格式如下：

[{'name': 'metaso_web_search',
  'title': None,
  'description': '根据关键词搜索网页、文档、论文、图片、视频、播客等内容',
  'inputSchema': {'type': 'object',
   'properties': {'q': {'description': '搜索查询关键词', 'type': 'string'},
    'size': {'description': '返回结果数量，默认10', 'type': 'integer'},
    'scope': {'description': '搜索范围：webpage, document, paper, image, video, podcast',
     'type': 'string'},
    'includeSummary': {'description': '是否包含摘要', 'type': 'boolean'},
    'includeRawContent': {'description': '是否包含原始内容', 'type': 'boolean'}},
   'required': ['q']},
  'outputSchema': None,
  'annotations': None,
  'meta': None},
 {'name': 'metaso_web_reader',
  'title': None,
  'description': '读取指定URL的网页内容',
  'inputSchema': {'type': 'object',
   'properties': {'format': {'description': '输出类型：json, markdown',
     'type': 'string'},
    'url': {'description': '要读取的URL地址', 'type': 'string'}},
   'required': ['format', 'url']},
  'outputSchema': None,
  'annotations': None,
  'meta': None},
 {'name': 'metaso_chat',
  'title': None,
  'description': '基于RAG的智能问答服务',
  'inputSchema': {'type': 'object',
   'properties': {'model': {'description': '使用的模型，默认fast', 'type': 'string'},
    'message': {'description': '用户问题', 'type': 'string'}},
   'required': ['message']},
  'outputSchema': None,
  'annotations': None,
  'meta': None}]

工具格式转换

此处需要一个转换函数，将MCP工具定义，转换成OpenAI API所能够接受的格式：

defconvert_tools(tools: List[Tool]):
"""
    将MCP Server的list tools获取到的工具列表，转换为OpenAI API的可用工具列表
    """
    ret = []
for tool in tools:
        parameters = {
"type": "object",
"properties": {},
"required": (
                tool.inputSchema["required"] if"required"in tool.inputSchema else []
            ),
        }
        properties = tool.inputSchema["properties"]
for param_name in properties:
if"type"in properties[param_name]:
                param_type = properties[param_name]["type"]
elif (
"anyOf"in properties[param_name]
and len(properties[param_name]["anyOf"]) > 0
            ):
                param_type = properties[param_name]["anyOf"][0]["type"]
else:
                param_type = "string"

            parameters["properties"][param_name] = {
"type": param_type,
            }
            parameters["properties"][param_name].update(properties[param_name])

        ret.append(
            {
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": parameters,
                },
            }
        )
return ret

转换后的工具格式如下：

[{'type': 'function',
'function': {'name': 'metaso_web_search',
'description': '根据关键词搜索网页、文档、论文、图片、视频、播客等内容',
'parameters': {'type': 'object',
'properties': {'q': {'type': 'string', 'description': '搜索查询关键词'},
'size': {'type': 'integer', 'description': '返回结果数量，默认10'},
'scope': {'type': 'string',
'description': '搜索范围：webpage, document, paper, image, video, podcast'},
'includeSummary': {'type': 'boolean', 'description': '是否包含摘要'},
'includeRawContent': {'type': 'boolean', 'description': '是否包含原始内容'}},
'required': ['q']}}},
 {'type': 'function',
'function': {'name': 'metaso_web_reader',
'description': '读取指定URL的网页内容',
'parameters': {'type': 'object',
'properties': {'format': {'type': 'string',
'description': '输出类型：json, markdown'},
'url': {'type': 'string', 'description': '要读取的URL地址'}},
'required': ['format', 'url']}}},
 {'type': 'function',
'function': {'name': 'metaso_chat',
'description': '基于RAG的智能问答服务',
'parameters': {'type': 'object',
'properties': {'model': {'type': 'string', 'description': '使用的模型，默认fast'},
'message': {'type': 'string', 'description': '用户问题'}},
'required': ['message']}}}]

调用工具

await mcp_client.acall_tool('metaso', 'metaso_web_search', {'q': '介绍一下DeepSeek-3.2-Exp', 'size': 4})

结果如下：

CallToolResult(meta=None, content=[TextContent(type='text', text='{"credits":3,"searchParameters":{"conciseSnippet":false,"format":"chat_completions","includeRawContent":false,"includeSummary":false,"q":"介绍一下DeepSeek-3.2-Exp","scope":"webpage","size":4},"total":27,"webpages":[{"authors":["黄心怡"],"date":"2025年09月29日","docId":"bfdf167f-bb4f-46e9-806d-af2daf9323f7","link":"https://finance.sina.com.cn/roll/2025-09-29/doc-infseiwf8257449.shtml","position":1,"score":"high","snippet":"!\\n[](https://n.sinaimg.cn/sinakd20250929s/125/w640h285/20250929/e229-6c4371c90d165930ef391cb1e279019a.png)\\n为了评估引入稀疏注意力机制的影响，DeepSeek方面特意将DeepSeek-V3.2-Exp的训练配置与V3.1-Terminus进行了对齐。\\n在各个领域的公开基准测试中，DeepSeek-V3.2-Exp的表现与V3.1-Terminus相当。\\n!\\n[](https://n.sinaimg.cn/sinakd20250929s/516/w640h676/20250929/cce0-0a6934e05c8471dedb3cffa2e3e3129b.png)\\nDeepSeek方面称，在新模型的研究过程中，需要设计和实现很多新的GPU算子。","title":"DeepSeek-V3.2-Exp 正式发布并开源"},{"date":"2025年09月29日","docId":"e17377bc-21d6-405a-8afb-9293671894b7","link":"https://www.sohu.com/a/939975717_121956424","position":2,"score":"high","snippet":"在人工智能和自然语言处理领域，技术的不断进步为我们带来了新的可能性。\\n今日（9月29日），DeepSeek正式发布了其全新的实验性版本——DeepSeek-V3.2-Exp。\\n这一版本不仅是对前一代V3.1-Terminus的更新，更是一次技术革新，标志着稀疏注意力机制在长文本处理上的新探索。\\n1. 深入了解DeepSeek-V3.2-Exp\\nDeepSeek-V3.2-Exp模型的推出，意味着在长文本的训练和推理效率方面，DeepSeek团队迈出了重要的一步。","title":"DeepSeek-V3.2-Exp"},{"date":"2025年09月29日","docId":"2aea422e-fe74-44bc-964c-80589fcff72e","link":"https://www.163.com/dy/article/KALDDBVM05566Y1D.html","position":3,"score":"high","snippet":"DeepSeek 果然又在过节搞事情了！\\n今天， DeepSeek-V3.2-Exp 版本正式发布。\\n该版本是基于公司此前发布的 DeepSeek-V3.1-Terminus 模型，升级而来。\\n版本命名中，Exp 是“实验版”的意思。\\n作为其下一代架构探索的关键中间步骤，新模型的核心亮点在于，引入了自主研发的 DeepSeek Sparse Attention (DSA) 稀疏注意力机制，以大幅优化长文本处理的训练和推理效率。","title":"DeepSeek-V3.2-Exp 版本正式发布"},{"authors":["陈骏达"],"date":"2023年09月30日","docId":"9a12ca13-1eb5-4c16-bfb7-8e08c4c413f3","link":"https://news.qq.com/rain/a/20250930A033C800","position":4,"score":"high","snippet":"结语：DeepSeek迈向新一代架构\\n正如其名字内的Exp（实验版）所言，DeepSeek-V3.2-Exp的推出，本身并不是一次性能爆表的升级，而更像是一场架构实验，展示了一种在长文本处理中兼顾性能和效率的新路径。\\n作为技术原型，DeepSeek-V3.2-Exp背后的DSA机制或许很快就会得到进一步完善。\\n随着相关技术的持续优化和更多企业、研究者参与验证，DeepSeek有望在不久的未来交出更令人惊喜的成果。","title":"DeepSeek离下一代架构，又近了一步！"}]}', annotations=None, meta=None, description=None)], structuredContent=None, isError=False, error=None)

可能大家都想学习AI大模型技术，也_想通过这项技能真正达到升职加薪，就业或是副业的目的，但是不知道该如何开始学习_，因为网上的资料太多太杂乱了，如果不能系统的学习就相当于是白学。
为了帮助大家打破壁垒，快速了解大模型核心技术原理，学习相关大模型技术。从原理出发真正入局大模型。在这里我和MoPaaS魔泊云联合梳理打造了系统大模型学习脉络，这份 LLM大模型资料 分享出来：包括LLM大模型书籍、640套大模型行业报告、LLM大模型学习视频、LLM大模型学习路线、开源大模型学习教程等, 😝有需要的小伙伴，可以 扫描下方二维码免费领取🆓**⬇️⬇️⬇️

在这里插入图片描述

【大模型全套视频教程】

教程从当下的市场现状和趋势出发，分析各个岗位人才需求，带你充分了解自身情况，get 到适合自己的 AI 大模型入门学习路线。

从基础的 prompt 工程入手，逐步深入到 Agents，其中更是详细介绍了 LLM 最重要的编程框架 LangChain。最后把微调与预训练进行了对比介绍与分析。

同时课程详细介绍了AI大模型技能图谱知识树，规划属于你自己的大模型学习路线，并且专门提前收集了大家对大模型常见的疑问，集中解答所有疑惑！

在这里插入图片描述

深耕 AI 领域技术专家带你快速入门大模型

跟着行业技术专家免费学习的机会非常难得，相信跟着学习下来能够对大模型有更加深刻的认知和理解，也能真正利用起大模型，从而“弯道超车”，实现职业跃迁！

在这里插入图片描述

【精选AI大模型权威PDF书籍/教程】

精心筛选的经典与前沿并重的电子书和教程合集，包含《深度学习》等一百多本书籍和讲义精要等材料。绝对是深入理解理论、夯实基础的不二之选。

在这里插入图片描述

【AI 大模型面试题 】

除了 AI 入门课程，我还给大家准备了非常全面的**「AI 大模型面试题」，**包括字节、腾讯等一线大厂的 AI 岗面经分享、LLMs、Transformer、RAG 面试真题等，帮你在面试大模型工作中更快一步。

【大厂 AI 岗位面经分享（92份）】

图片

【AI 大模型面试真题（102 道）】

图片

【LLMs 面试真题（97 道）】

图片

【640套 AI 大模型行业研究报告】

在这里插入图片描述

【AI大模型完整版学习路线图（2025版）】

明确学习方向，2025年 AI 要学什么，这一张图就够了！

img

👇👇点击下方卡片链接免费领取全部内容👇👇

在这里插入图片描述

抓住AI浪潮，重塑职业未来！

科技行业正处于深刻变革之中。英特尔等巨头近期进行结构性调整，缩减部分传统岗位，同时AI相关技术岗位（尤其是大模型方向）需求激增，已成为不争的事实。具备相关技能的人才在就业市场上正变得炙手可热。

行业趋势洞察：

  • 转型加速： 传统IT岗位面临转型压力，拥抱AI技术成为关键。
  • 人才争夺战： 拥有3-5年经验、扎实AI技术功底和真实项目经验的工程师，在头部大厂及明星AI企业中的薪资竞争力显著提升（部分核心岗位可达较高水平）。
  • 门槛提高： “具备AI项目实操经验”正迅速成为简历筛选的重要标准，预计未来1-2年将成为普遍门槛。

与其观望，不如行动！

面对变革，主动学习、提升技能才是应对之道。掌握AI大模型核心原理、主流应用技术与项目实战经验，是抓住时代机遇、实现职业跃迁的关键一步。

在这里插入图片描述

01 为什么分享这份学习资料？

当前，我国在AI大模型领域的高质量人才供给仍显不足，行业亟需更多有志于此的专业力量加入。

因此，我们决定将这份精心整理的AI大模型学习资料，无偿分享给每一位真心渴望进入这个领域、愿意投入学习的伙伴！

我们希望能为你的学习之路提供一份助力。如果在学习过程中遇到技术问题，也欢迎交流探讨，我们乐于分享所知。

*02 这份资料的价值在哪里？*

专业背书，系统构建：

  • 本资料由我与MoPaaS魔泊云的鲁为民博士共同整理。鲁博士拥有清华大学学士和美国加州理工学院博士学位，在人工智能领域造诣深厚：

    • 在IEEE Transactions等顶级学术期刊及国际会议发表论文超过50篇。
    • 拥有多项中美发明专利。
    • 荣获吴文俊人工智能科学技术奖（中国人工智能领域重要奖项）。

  • 目前，我有幸与鲁博士共同进行人工智能相关研究。

在这里插入图片描述

内容实用，循序渐进：

  • 资料体系化覆盖了从基础概念入门到核心技术进阶的知识点。

  • 包含丰富的视频教程与实战项目案例，强调动手实践能力。

  • 无论你是初探AI领域的新手，还是已有一定技术基础希望深入大模型的学习者，这份资料都能为你提供系统性的学习路径和宝贵的实践参考，助力你提升技术能力，向大模型相关岗位转型发展。

    在这里插入图片描述在这里插入图片描述在这里插入图片描述

抓住机遇，开启你的AI学习之旅！

在这里插入图片描述

# 数据库 # 人工智能 # 大数据 # 自然语言处理 # 语言模型
Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

cover

n8n智能体开发：Azure OpenAI 聊天模型节点

avatar 2048 AI社区
cover

n8n智能体开发：AWS Bedrock Chat Model 节点

avatar 2048 AI社区

超越 FOMO——在人工智能领域保持最新动态

原文：towardsdatascience.com/beyond-fomo-keeping-up-to-date-in-ai-8b22b7050cdf?

avatar 2048 AI社区