最近研究了下MCP，一个比较混乱的技术，所有介绍MCP的文章都会讲到它的文档令人费解，我也折腾了很久。

第一次看MCP，我就想快速搞明白输入、输出，但是MCP的文档里很奇怪的一点不讲这一点，比如stdio方式运行的MCP， 是一个进程，那么它的输入是什么？输出是什么？如何交互，我能否观察他的输入输出？能否通过外部往这个进程写入？

最后我也放弃了这些，把它当作黑盒子用吧。

现在实现了一个简单的MCP客户端，通过类似cursor的MCP.json配置，实现了两个核心功能，list tools, call tool 每一个都是单独的一次命令调用或者http请求，未实现状态维护，我还没有发现这方面的用例。

list tools

MCP向大模型暴露了一些工具列表，包括调用参数等。（注意这里并不需要定义返回值）

call tool

  1. 大模型返回工具调用的指令
  2. 本地客户 端向MCP服务器发起工具调用请求，使用call tool
  3. MCP服务器返回调用结果，常为文本
  4. 本地客户端将调用结果返回给大模型，我发现使用json字符串即可，并不需要转化为markdown之类的格式，大模型理解json没有问题。

问题

windows下，打印异常的问题

下面的代码，在windows下，会打印异常信息，可能和windows下进程资源清理有关。

import asyncio
import contextlib
from contextlib import suppress
from mcp import ClientSession, StdioServerParameters, stdio_client
import os
import sys


@contextlib.contextmanager
def suppress_stderr():
    """临时重定向 stderr 到 null 设备"""
    original_stderr = sys.stderr
    try:
        with open(os.devnull, 'w') as devnull:
            sys.stderr = devnull
            yield
    finally:
        sys.stderr = original_stderr

async def main():
    server_params = StdioServerParameters(
        command="npx", args=["-y", "@playwright/mcp@latest"]
    ) # Or any other server
    
    # 同时抑制 stderr 并捕获 ValueError
    with suppress_stderr(), contextlib.suppress(ValueError):
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                server_tools = await session.list_tools()
                tools = server_tools.model_dump().get("tools", [])
                print(tools[:3])
                await asyncio.sleep(2.0)

if __name__ == "__main__":
    asyncio.run(main())

修复

    with suppress_stderr(), contextlib.suppress(ValueError):
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                server_tools = await session.list_tools()
                tools = server_tools.model_dump().get("tools", [])
                print(tools[:3])
    await asyncio.sleep(2.0) # <---这里加延时