MCP传输：让JSON消息跨进程交换

mkdir -Force "F:\学习\agentdepth-l2-mcp-transport"
cd "F:\学习\agentdepth-l2-mcp-transport"

"""通过标准输入接收消息，分别输出协议响应与诊断日志。"""

import json
import sys


# 依据方法名构造响应，沿用收到的请求编号。
def reply(message):
    request_id = message.get("id")
    method = message.get("method")
    if method == "initialize":
        return {"jsonrpc": "2.0", "id": request_id, "result": {
            "protocolVersion": "2025-11-25", "capabilities": {"tools": {}},
            "serverInfo": {"name": "stdio-demo", "version": "0.1.0"}}}
    if method == "tools/list":
        return {"jsonrpc": "2.0", "id": request_id, "result": {"tools": []}}
    return {"jsonrpc": "2.0", "id": request_id,
            "error": {"code": -32601, "message": "Method not found"}}


def main():
    initialized = False
    ready = False
    for raw in sys.stdin:
        try:
            message = json.loads(raw)
            if not isinstance(message, dict):
                raise ValueError("object_required")
        except (json.JSONDecodeError, ValueError):
            print("invalid_input", file=sys.stderr)
            continue
        method = message.get("method")
        if method == "notifications/initialized":
            if initialized:
                ready = True
                print("initialized_notification", file=sys.stderr)
            else:
                print("notification_before_initialize", file=sys.stderr)
            continue
        if method == "initialize":
            initialized = True
            response = reply(message)
        elif method == "tools/list" and not ready:
            response = {"jsonrpc": "2.0", "id": message.get("id"),
                        "error": {"code": -32000, "message": "Not initialized"}}
        else:
            response = reply(message)
        print(json.dumps(response, ensure_ascii=False), flush=True)
        print("handled_method=" + str(method), file=sys.stderr)


if __name__ == "__main__":
    main()

"""启动子进程，逐次发送消息，并分别读取响应与日志。"""
import json
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

def exchange():
    server = Path(__file__).with_name("stdio_server.py")
    process = subprocess.Popen([sys.executable, str(server)],
        stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
        text=True, encoding="utf-8")
    pool = ThreadPoolExecutor(max_workers=1)
    def send(message):
        # 一行一个JSON消息，flush让服务端及时收到。
        process.stdin.write(json.dumps(message, ensure_ascii=False) + "\n")
        process.stdin.flush()
    def receive(expected_id):
        # 在线程中等待一行，调用方最多等待5秒。
        line = pool.submit(process.stdout.readline).result(timeout=5)
        response = json.loads(line)
        if response.get("id") != expected_id or "error" in response:
            raise ValueError("响应编号或状态异常")
        return response
    try:
        send({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {
            "protocolVersion": "2025-11-25", "capabilities": {},
            "clientInfo": {"name": "stdio-client-demo", "version": "0.1.0"}}})
        initialized = receive(1)
        if initialized["result"].get("protocolVersion") != "2025-11-25":
            raise ValueError("服务端协议版本与当前客户端不匹配")
        # 收到并核对初始化响应后，再发送准备通知和目录请求。
        send({"jsonrpc": "2.0", "method": "notifications/initialized"})
        send({"jsonrpc": "2.0", "id": 2, "method": "tools/list"})
        catalog = receive(2)
        process.stdin.close()
        if process.wait(timeout=5) != 0:
            raise RuntimeError("服务进程退出异常")
        return [initialized, catalog], process.stderr.read().splitlines()
    finally:
        if process.poll() is None:
            process.kill()
            process.wait(timeout=5)
        pool.shutdown(wait=True)
        for stream in (process.stdin, process.stdout, process.stderr):
            if hasattr(stream, "close"):
                stream.close()

if __name__ == "__main__":
    result, diagnostics = exchange()
    print("response_ids", [item["id"] for item in result])
    print("tools", result[-1]["result"]["tools"])
    print("stderr_events", diagnostics)

python stdio_client.py

response_ids [1, 2]
tools []
stderr_events ['handled_method=initialize', 'initialized_notification', 'handled_method=tools/list']

python -m unittest test_stdio_demo -v