Python连接MCP Server全流程：初始化、工具调用与远程鉴权实战-嵌云网-嵌入式AI开发资源站

Python连接MCP Server全流程：初始化、工具调用与远程鉴权实战

崔怂包·2026-10-10·阅读 22 GEO检测（免费）

MCPMCP ServerPython

|难度：普通|阅读约 39 分钟

看到这个标题，我第一反应是：MCP 这个词最近在 AI 圈子里火得不行，但真正能把它跑通的人其实没那么多。很多朋友卡在“理论看了一堆，连第一个 MCP Server 都没连上”的阶段。我结合自己这一阵子实际折腾 Python 连接 MCP Server 的经验，把从零开始的完整路径走了一遍，包括客户端初始化、工具调用、远程连接、鉴权，以及那些文档里不会明说的坑。这篇就按我实操的顺序来写，代码直接可复现。

MCP Server 说白了就是一个“给大模型用的工具接口层”。它把数据库、文件系统、外部 API、内部系统这些能力，统一打包成标准化的工具，让大模型通过一个固定协议去调用。Python 开发者在这套体系里的角色很特殊：我们既能写 Server 端，也能写 Client 端。本文重点讲 Client 端，也就是用 Python 去主动连接一个已经存在的 MCP Server，列出它提供了哪些工具，然后调用这些工具，拿到结构化结果。适合正在做 AI Agent 集成、想把自己的数据源暴露给大模型、或者想在公司内部搭建工具网关的开发者参考。

1. 先弄清 MCP Server 到底是什么

1.1 从一个调用场景说起

假设你做了一个内部运维助手，想让大模型帮忙查服务器状态。如果没有 MCP，你得在提示词里塞一堆 JSON 格式说明，告诉模型“什么时候该调用哪个 HTTP 接口，参数怎么拼”。等接口数量超过五六个，这种方法就会崩盘：模型记不住、参数经常写错、接口返回格式五花八门，解析逻辑变成一坨乱麻。

MCP 出现以后，思路彻底变了。Server 端把“查服务器状态”封装成一个名为 get_server_status 的工具，声明好参数结构，然后交给模型。模型决定调用时，会按这个结构生成参数，你的 Python Client 负责把请求发出去，再把结果塞回给模型。整个过程是标准化的，跟具体业务无关。这就是 MCP 的核心价值：把工具描述、参数校验、结果返回这三件事统一成一种协议。

很多人混淆一个概念：MCP 不是 HTTP API，也不是 RPC 框架。它更接近一种“协议+运行时规范”的组合。底层你可以用 stdio 传输，也可以用 SSE、streamable HTTP 传输，但上面跑的会话语义是一样的。也就是说，MCP 定义了“客户端和服务端怎么对话”，但不强制“对话用什么通道”，这个设计直接影响了我们后续选型。

1.2 Python 在 MCP 生态中的位置

Python 在 MCP 生态里几乎算是一等公民。官方 SDK 最早提供的就是 Python 和 TypeScript 两个版本，社区里大量现成 Server 也是 Python 写的，比如给开发工具用的文件访问 Server、给数据分析用的数据库查询 Server。更关键的是，Python 的异步生态和 MCP 的异步会话模型天然匹配。

MCP 的会话是异步双向的：客户端发送初始化请求，服务端推送工具列表和调用结果，中间还可能穿插日志和错误通知。如果用过 WebSocket 之类的协议，你会觉得很熟悉。用 Python 的 asyncio 处理这种双向流非常顺手，官方 SDK 也是围绕 asyncio 构建的。

1.3 传输方式：stdio 与远程协议

连接 MCP Server，首先得选传输方式。

stdio 模式：最常用于本地工具。Client 用 subprocess 启动 Server 进程，然后通过进程的标准输入输出进行 JSON-RPC 消息交换。典型场景是编辑器插件：扩展启动一个本地 Python 脚本作为 MCP Server，大模型通过这个脚本访问本地文件。优点是零网络配置、进程隔离好；缺点是 Server 必须和 Client 在同一台机器上。

SSE 模式：用于远程 Server。Client 通过 HTTP POST 发送请求，同时通过一个长连接的 SSE（Server-Sent Events）流接收服务端消息。典型场景是连接一个部署在云上的工具服务。SSE 在早期版本里支持得比较多，后来规范逐渐向 streamable HTTP 演进。实操中你会发现，很多现成 SDK 对 SSE 和 streamable HTTP 的封装层级不一样，接口参数略有差异。

其他模式：2025 年之后规范不断更新，还出现了 streamable HTTP 等新传输。我的建议是：本地调试用 stdio，生产远程连接优先看 Server 支持哪种协议，不要盲目追求某个特定模式。连接成功的关键是理解“Client 和 Server 必须配合同一种传输”，两边不一致必然失败。

需要模型API调用？免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。

2. 动手前的准备工作

2.1 Python 版本与虚拟环境

先说结论：推荐 Python 3.10 及以上。MCP SDK 使用了不少较新的类型语法和异步特性，3.9 以下会有兼容问题。我自己先在 3.8 环境试过一次，光类型注解就报了一片，后来直接切到 3.11。

创建虚拟环境是老生常谈，但我还是想强调一次。MCP SDK 的依赖不算重，但它会带动 httpx、pydantic、anyio 这些关键库。如果直接装到全局环境，很容易和项目里其他依赖冲突。别偷懒，执行这几步：

bash复制python3.11 -m venv .venv
source .venv/bin/activate
pip install --upgrade pip

2.2 安装 MCP SDK

官方 Python SDK 包名就叫 mcp，安装命令很简单：

bash复制pip install mcp

但这里有个隐藏信息：mcp 现在是个命名空间包，你可能还会看到 mcp[cli] 这种扩展安装方式。如果只做客户端连接，裸装 mcp 就够。如果还想自己快速编写 Server 并调试，建议直接安装完整版：

bash复制pip install "mcp[cli]"

安装完成后，可以跑一下验证：

bash复制python -c "import mcp; print(mcp.__version__)"

能打印出版本号，说明核心库没问题。我第一次安装后卡在这一步很久，后来发现是 pip 缓存了旧版本，加了 --upgrade 重装才解决。

2.3 从哪找一个能连的 Server

如果你手头没有现成的 MCP Server，建议先自己写一个最简单的来练手。用官方 SDK 里的 FastMCP 可以极快地搭建一个：

python复制# my_mcp_server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("DemoServer")

@mcp.tool()
def add(a: int, b: int) -> int:
    """计算两个整数之和"""
    return a + b

if __name__ == "__main__":
    mcp.run()

保存后，直接用下面的客户端代码去连它。这样你就能在一个完全可控的环境里搞懂连接逻辑，不用去依赖外部服务，排查问题也方便得多。等熟悉之后再连接真正的远程业务 Server，会顺手很多。

3. 核心实操：用 Python 客户端连上一个 MCP Server

3.1 最简流程：初始化会话并列出工具

连接过程的核心 API 有三个：stdio_client、ClientSession、session.initialize()。我直接贴一段最简可用的代码：

python复制import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    # 指定本地 Server 的启动方式
    server_params = StdioServerParameters(
        command="python",
        args=["my_mcp_server.py"],
    )

    # 启动子进程，建立双向流
    async with stdio_client(server_params) as (read_stream, write_stream):
        # 基于流创建会话
        async with ClientSession(read_stream, write_stream) as session:
            # 第一步永远是 initialize
            await session.initialize()

            # 列出 Server 提供的所有工具
            tools_result = await session.list_tools()
            for tool in tools_result.tools:
                print(f"工具名: {tool.name}")
                print(f"描述: {tool.description}")
                print(f"输入结构: {tool.inputSchema}")
                print("---")

asyncio.run(main())

这里有几个关键点必须强调。

第一，stdio_client 返回的是一个元组 (read_stream, write_stream)。这是最容易被忽略的细节。很多人以为它返回一个流对象，直接拿 async with 去套，结果报错。官方文档里的写法就是解包成两个流，我建议直接照抄，别自创。

第二，ClientSession 的初始化必须显式调用 initialize()。这是我踩过的最大一个坑。如果忘记调用，后续 list_tools 和 call_tool 会抛出“session not initialized”之类的错误。原因在于 MCP 协议规定双方的握手动作是必须的，SDK 不会自动替你完成。很多所谓“连接失败”的报错，本质上都是少了这一步。

第三，整个流程必须放在异步上下文管理器里。stdio_client 启动的是一个真正的子进程，如果不用 async with 管理，子进程的生命周期就没法保证，连接结束以后可能会留下僵尸进程。我在开发时偶尔发现本地多了一堆 Python 子进程，排查后发现就是连接异常退出后没有清理干净。

3.2 真正调用工具并处理返回值

列出工具只是第一步，真正干活的是 call_tool。先看代码：

python复制import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    server_params = StdioServerParameters(
        command="python",
        args=["my_mcp_server.py"],
    )

    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            result = await session.call_tool(
                name="add",
                arguments={"a": 10, "b": 32},
            )

            print("类型:", result.type)          # 通常是 "text"
            print("原始内容:", result.content)
            print("是否报错:", result.isError)

            # 提取纯文本
            for item in result.content:
                if item.type == "text":
                    print("返回文本:", item.text)

asyncio.run(main())

如果顺利，你会看到终端输出类似于“返回文本: 42”。注意 call_tool 的返回值是 CallToolResult，它的结构跟很多人想的不一样：不是直接返回 JSON，而是返回一个包含 content 列表的对象。content 里每个元素有 type 字段，最常见的 text 类型，下面还有 text 字段。如果 Server 返回结构化数据，你还需要进一步解析。

这里必须提醒一点：isError 字段不是 Python 异常。MCP 协议里，工具即使执行失败，返回的也只是一个带有 isError=True 的结果对象，而不是抛出异常。如果你用 try-except 去捕获，大概率捕获不到。正确做法是调用以后主动检查 result.isError，再决定后续处理。这个设计初看别扭，但想通就明白了：服务端在执行工具时遇到了业务错误，它仍然要把这个“错误结果”传回给模型，而不是中断整个会话。

3.3 把连接逻辑封装成可复用客户端

每次写一遍 stdio_client 和 initialize 太繁琐。我通常会封装成一个类，方便在多个项目里复用：

python复制import asyncio
from typing import Any
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


class MCPClient:
    def __init__(self, command: str, args: list[str]):
        self.server_params = StdioServerParameters(command=command, args=args)
        self.session: ClientSession | None = None
        self._stream_ctx = None

    async def __aenter__(self):
        self._stream_ctx = stdio_client(self.server_params)
        read_stream, write_stream = await self._stream_ctx.__aenter__()
        self.session = await ClientSession(read_stream, write_stream).__aenter__()
        await self.session.initialize()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        if self.session:
            await self.session.__aexit__(exc_type, exc, tb)
        if self._stream_ctx:
            await self._stream_ctx.__aexit__(exc_type, exc, tb)

    async def list_tools(self):
        return await self.session.list_tools()

    async def call_tool(self, name: str, arguments: dict[str, Any]):
        result = await self.session.call_tool(name, arguments)
        if result.isError:
            raise RuntimeError(f"工具 {name} 执行失败: {result.content}")
        return result


async def main():
    async with MCPClient(command="python", args=["my_mcp_server.py"]) as client:
        tools = await client.list_tools()
        print([t.name for t in tools.tools])

        result = await client.call_tool("add", {"a": 2, "b": 3})
        print(result.content[0].text)

asyncio.run(main())

封装之后，核心连接细节被隐藏了，调用方只需要关注业务参数。不过这里有个额外经验：不要在这个封装里过早假设 result.isError 一定代表需要抛异常。有些场景下，你恰恰希望把错误结果原样返回给上层模型（例如让模型自行调整参数重试）。所以我会提供一个 raw_call_tool 方法，把原始结果暴露出来，而 call_tool 则是带着业务判断的封装版。

4. 连接远程 MCP Server 与鉴权处理

4.1 通过 SSE 连接远程服务

本地 Server 用 stdio 没问题，但生产环境里 Server 多半部署在远程。这时要换成 sse_client。代码结构和 stdio 很像，只是参数从命令换成了 URL：

python复制import asyncio
from mcp import ClientSession
from mcp.client.sse import sse_client

async def main():
    url = "https://your-remote-server.example/mcp"
    async with sse_client(url) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            tools = await session.list_tools()
            print([tool.name for tool in tools.tools])

asyncio.run(main())

看起来跟 stdio 版本几乎一样，但底层机制完全不同。sse_client 会做两件事：POST 到指定 URL 发起会话，然后建立一个 SSE 长连接接收服务端消息。如果你的 Server 是旧版协议，URL 可能不是一个根地址，而是分成了“endpoint”和“message endpoint”两个。新版 streamable HTTP 协议则收敛成一个地址。所以我建议连接前先确认 Server 支持的协议版本和确切 endpoint，否则就会遇到 404 或 405 之类的错误。

4.2 带 Token 鉴权的连接封装

远程 Server 不可能是裸奔的，一般都会要求鉴权。MCP 协议本身没有定义鉴权方式，鉴权落在传输层。SSE/HTTP 场景通常用 Authorization 请求头。

官方 SDK 里，sse_client 支持传入 headers 参数，但具体签名会随版本变化。我实测下来比较稳妥的写法是：

python复制import asyncio
from mcp import ClientSession
from mcp.client.sse import sse_client

async def main():
    headers = {
        "Authorization": "Bearer your-token-here",
        "X-Custom-Tenant": "tenant-a",
    }
    url = "https://your-remote-server.example/mcp"

    async with sse_client(url, headers=headers) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            result = await session.call_tool(
                name="query_order",
                arguments={"order_id": "12345"},
            )
            print(result.content[0].text)

asyncio.run(main())

有个容易踩的坑：某些封装版本里，headers 参数名字可能不同，也可能要求把所有鉴权信息拼进 URL。我建议拿到 SDK 之后先 inspect.signature(sse_client) 看一下参数列表。另外，绝对不要把 Token 硬编码在代码里，至少用环境变量：

python复制import os

headers = {
    "Authorization": f"Bearer {os.environ.get('MCP_API_TOKEN', '')}"
}

还有一点，SSE 模式下你可能还需要区分“服务端的 endpoint 和客户端收消息的 endpoint”。有些部署会把它们分成两个 URL，比如 https://host/mcp/sse 和 https://host/mcp/messages。如果你只配置了一个 URL 导致握手失败，看看 Server 文档里有没有两个 endpoint 的说明。

4.3 连接超时与重试策略

远程连接最恶心的就是网络抖动。SDK 底层用了 httpx，默认超时可能不够用。特别是 SSE 长连接，如果服务端 30 秒没有消息，客户端可能会因为读超时直接断开。我实测下来，有两种处理思路。

思路一：在创建连接前，给底层的 httpx 客户端设置更大的超时。有些版本的 sse_client 支持传入 timeout 参数，单位是秒，可以设置成 (连接超时, 读超时) 这种元组。

思路二：自己封装重试逻辑。比如第一次连接失败，退避 1 秒重试，最多三次：

python复制import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=5))
async def connect_with_retry(url, headers):
    from mcp.client.sse import sse_client
    return await sse_client(url, headers=headers).__aenter__()

不过要注意：重试的对象最好是“建立连接”这一步，而不是“整个会话”。会话一旦建立，再重试就可能导致状态不一致。如果连接成功后工具调用超时，我倾向于把具体调用重试而不是重建会话，这需要更细致的代码设计。

5. 常见问题与排查技巧实录

5.1 子进程闪退导致连接立刻断开

这也是 stdio 模式最常见的问题。表现是：代码一进入 async with stdio_client 就直接异常，或者会话建立后立即报错“管道已关闭”。排查办法很简单，先在终端手动跑一遍命令：

bash复制python my_mcp_server.py

如果这个命令本身有语法错误、未捕获异常，子进程启动后会闪退。MCP 客户端这边感知到的就是“读不到任何消息，连接关闭”。还有一个隐蔽原因：Server 脚本里如果有 if __name__ == "__main__": 分支之外过早退出逻辑，也会导致同样问题。

我的排查经验是：在 stdio_client 参数里不要用 command="python" 字符串拼接，直接用 sys.executable 会更稳妥，避免虚拟环境不对：

python复制import sys
from mcp import StdioServerParameters

server_params = StdioServerParameters(
    command=sys.executable,
    args=["my_mcp_server.py"],
)

这样客户端进程使用的解释器就是当前虚拟环境的解释器，不会跑到系统全局 Python 里。

5.2 异步环境冲突：线程、Jupyter 与事件循环

很多人第一次跑示例代码，把 asyncio.run(main()) 直接扔进 Jupyter Notebook 里，结果报错“This event loop is already running”。因为 Jupyter 内部已经启动了一个事件循环，再调 asyncio.run() 会冲突。

解决办法有几个，我按推荐排序：

用 await main()（Jupyter 支持直接 await）。
在独立 .py 脚本里用 asyncio.run()。
在不能避免嵌套的场景用 nest_asyncio 补丁：

python复制import nest_asyncio
nest_asyncio.apply()

另一个坑是：如果你在多线程环境中调用异步连接（比如在 FastAPI 的线程池里），必须确保同一个事件循环管理整个会话。MCP Client 一旦在某个事件循环上创建，就不要把这个 session 对象扔到另一个线程去调用。别问我怎么知道的，把 session 对象当作线程变量传递是我踩过最深的坑之一。

5.3 SDK 版本差异导致的 API 变化

MCP 规范还在快速演进，SDK 的 API 变化也快。我遇到过几种典型变化：sse_client 的返回从 (read_stream, write_stream) 变成带上下文的对象；StdioServerParameters 的字段改名；list_tools 返回结构从 tools 列表变成 result.tools 带额外元数据。

应对方法是：升级 SDK 前先看 changelog，同事之间协作时锁定版本。可以在 requirements.txt 里固定版本号，例如：

code复制mcp==1.*

而不是直接 mcp>=1 这样粗放。如果不幸遇到旧代码在新 SDK 上跑不了，优先考虑回退版本而不是立即改造代码。因为这个库迭代太快，今天改造完，明天可能又变了。我自己的经验是：锁定一个稳定版本，然后研究透它，比频繁追新更高效。

5.4 工具返回内容解析错误

call_tool 返回的 content 列表，并不总是单纯的文本。有些 Server 会返回 JSON 结构、图片路径、资源引用等类型。我见过很多新手一上来就写 result.content[0].text，结果 Server 返回的是带 type="image" 的内容，直接崩溃。

正确做法是先检查 item.type 再解析：

python复制for item in result.content:
    if item.type == "text":
        print(item.text)
    elif item.type == "image":
        print("图片数据，可能 base64:", item.data[:100])
    elif item.type == "resource":
        print("资源引用:", item.resource)

另外，即使 type == "text"，文本内部也可能是 JSON 字符串，需要二次 json.loads。很多 Server 的 inputSchema 里声明了返回格式，但实际实现并不严格，宁可多一步解析，也不要裸打印就完事。

5.5 日志看不到、错误被吞

stdio 模式下，Server 进程的 stdout 被 MCP 协议占用了，Server 里用 print() 打日志会出现两种情况：要么被协议解析器当成非法消息导致连接中断，要么直接丢失。很多人在 Server 端加了一堆 print 调试，结果发现客户端啥也没收到。

正确做法是让 Server 把日志写到 stderr：

python复制import sys

def log(msg: str):
    print(msg, file=sys.stderr)

客户端这边，stdio_client 通常会透传子进程的 stderr，但未必打到你看到的终端。我建议在启动 Server 的命令里加上 env 参数，把子进程的标准错误也重定向到当前进程：

python复制import os
import sys
from mcp import StdioServerParameters

server_params = StdioServerParameters(
    command=sys.executable,
    args=["my_mcp_server.py"],
    env={**os.environ, "PYTHONUNBUFFERED": "1"},
)

PYTHONUNBUFFERED=1 可以使 Python 子进程的日志不经过缓冲，及时输出，排查问题的时候特别有用。

写在最后的经验

MCP 的连接链路我第一次跑通花了大半天，真正写业务代码时间反而更久。前期卡壳主要是不熟悉异步上下文、遗漏了 initialize()，以及搞混了 stdio 与 SSE 的适用场景。如果你正在入门，建议一定先用自己的 FastMCP Server 跑通最简单的“列表-调用-返回”闭环，再考虑接远程业务。多留一点时间给 SDK 版本兼容和日志排查相关的代码，它们在实际环境中几乎必然会出问题。这个协议还在快速迭代，今天的结论过几个月未必完全适用，但“先把最小闭环跑通”这件事，任何版本都绕不开。