MCP 实战（三）MCP 客户端

前面我们讲解了 AI 客户端的能力扩展方式。对于 AI 应用开发者而言，我们可以在项目中集成 MCP 客户端，自动探测 MCP 服务端暴露的所有工具，并根据自身业务逻辑，灵活发起工具调用。

下面提供两套最常用的 MCP 客户端极简示例，分别为 stdio 本地进程模式 和 streamable-http 网络请求模式，完整演示连接服务 → 发现工具 → 调用工具的完整流程。

import asyncio

# 1. stdio 客户端
from mcp.client.stdio import stdio_client
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters

async def main1():
    # 启动 MCP 服务端子进程，拿到和子进程进行通信的异步读写流 reader、writer
    async with stdio_client(StdioServerParameters(command='python', args=['mcp_server.py',])) as (reader, writer):
        # 创建 MCP 客户端会话，基于读写流封装 MCP 协议 RPC 调用
        async with ClientSession(reader, writer) as session:
            await session.initialize()

            # 在主进程获取服务端暴露的工具列表
            res = await session.list_tools()
            print(res.tools)

            # 在主进程发起工具调用 RPC 请求，工具逻辑实际运行在子进程
            ret = await session.call_tool(name='get_weather', arguments={'loc': '北京'})
            print('工具调用结果:', ret.content)



# 2. streamable-http 客户端
from mcp.client.streamable_http import streamable_http_client
async def main2():
    async with streamable_http_client(url='http://127.0.0.1:8000/mcp') as (reader, writer):
        async with ClientSession(reader, writer) as session:
            await session.initialize()

            res = await session.list_tools()
            print(res.tools)

            ret = await session.call_tool(name='get_weather', arguments={'loc': '北京'})
            print('工具调用结果:', ret.content)



if __name__ == '__main__':
    # asyncio.run(main1())
    asyncio.run(main2())