Первый MCP-сервер на Python: от пустой папки до ответа инструмента

py -3.11 --version
$labPath = Join-Path $env:USERPROFILE 'mcp-label-lab'
New-Item -ItemType Directory -Path $labPath -ErrorAction Stop
Set-Location -LiteralPath $labPath
py -3.11 -m venv .venv

mcp==1.30.0

.\.venv\Scripts\python.exe -m pip install -r requirements-label.txt
.\.venv\Scripts\python.exe -c "from importlib.metadata import version; print(version('mcp'))"

"""Local stdio tutorial for mcp==1.30.0. No external services or keys."""
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("label-lab")


@mcp.tool()
def normalize_label(text: str) -> dict[str, str]:
    """Trim a label and reject empty labels or more than 80 characters."""
    label = text.strip()
    if not 1 <= len(label) <= 80:
        raise ValueError("Expected 1 to 80 characters after trimming")
    return {"label": label}


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

"""Call the local tutorial via JSON-RPC/stdin, then stop its child process."""
import json
import os
from pathlib import Path
import queue
import subprocess
import sys
import threading


def main() -> dict:
    directory = Path(__file__).resolve().parent
    server = directory / "label_server.py"
    if not server.is_file():
        raise FileNotFoundError("Save label_server.py beside check_stdio.py")
    process = subprocess.Popen(
        [sys.executable, "-u", str(server)],
        cwd=directory,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        encoding="utf-8",
        bufsize=1,
        creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0,
    )
    replies = queue.Queue()
    diagnostics = []

    def read_stdout():
        for line in process.stdout:
            replies.put(line)
        replies.put(None)

    def read_stderr():
        for line in process.stderr:
            diagnostics.append(line.rstrip())

    threading.Thread(target=read_stdout, daemon=True).start()
    threading.Thread(target=read_stderr, daemon=True).start()
    transcript = []

    def send(message):
        transcript.append({"direction": "client", "message": message})
        process.stdin.write(json.dumps(message) + "\n")
        process.stdin.flush()

    def request(identifier, method, params):
        send({"jsonrpc": "2.0", "id": identifier, "method": method, "params": params})
        while True:
            line = replies.get(timeout=10)
            if line is None:
                raise RuntimeError("Server stopped: " + " | ".join(diagnostics[-4:]))
            message = json.loads(line)
            transcript.append({"direction": "server", "message": message})
            if message.get("id") == identifier:
                return message

    try:
        initialized = request(1, "initialize", {
            "protocolVersion": "2025-11-25",
            "capabilities": {},
            "clientInfo": {"name": "label-lab-check", "version": "1.0.0"},
        })
        if initialized.get("result", {}).get("protocolVersion") != "2025-11-25":
            raise AssertionError("Unexpected negotiated protocol version")
        send({"jsonrpc": "2.0", "method": "notifications/initialized"})
        listing = request(2, "tools/list", {})
        tools = listing.get("result", {}).get("tools", [])
        if [tool["name"] for tool in tools] != ["normalize_label"]:
            raise AssertionError("normalize_label is missing from tools/list")
        schema = tools[0]["inputSchema"]
        if schema["properties"]["text"]["type"] != "string" or "text" not in schema["required"]:
            raise AssertionError("Tool input schema changed")
        success = request(3, "tools/call", {"name": "normalize_label", "arguments": {"text": "  Draft  "}})
        result = success.get("result", {})
        if result.get("isError") or result.get("structuredContent") != {"label": "Draft"}:
            raise AssertionError("Unexpected normalized label")
        cases = [(4, "blank", {"text": "   "}), (5, "too_long", {"text": "x" * 81}), (6, "missing_text", {})]
        errors = {}
        for identifier, name, arguments in cases:
            reply = request(identifier, "tools/call", {"name": "normalize_label", "arguments": arguments})
            if reply.get("result", {}).get("isError") is not True:
                raise AssertionError("Expected an error for " + name)
            errors[name] = reply["result"]
        return {"passed": True, "protocol": initialized["result"]["protocolVersion"], "tool": tools[0], "success": result, "known_errors": errors, "transcript": transcript}
    finally:
        process.stdin.close()
        try:
            process.wait(timeout=3)
        except subprocess.TimeoutExpired:
            process.terminate()
            try:
                process.wait(timeout=3)
            except subprocess.TimeoutExpired:
                process.kill()
                process.wait(timeout=3)


if __name__ == "__main__":
    try:
        report = main()
        print("PASS: initialize, tools/list, tools/call, 3 expected errors")
        print(json.dumps(report, ensure_ascii=False, indent=2))
    except Exception as error:
        print("FAIL:", error, file=sys.stderr)
        raise SystemExit(1)

.\.venv\Scripts\python.exe check_stdio.py

PASS: initialize, tools/list, tools/call, 3 expected errors

{
  "content": [
    {
      "type": "text",
      "text": "{\n  \"label\": \"Draft\"\n}"
    }
  ],
  "structuredContent": {
    "label": "Draft"
  },
  "isError": false
}

"""Create this new lab's portable MCP configuration with absolute paths."""
import json
from pathlib import Path
import sys

directory = Path(__file__).resolve().parent
configuration = directory / ".mcp.json"
if configuration.exists():
    raise SystemExit(".mcp.json already exists; keep it and review it manually")
server = directory / "label_server.py"
if not server.is_file():
    raise SystemExit("Save label_server.py beside configure_client.py")
configuration.write_text(json.dumps({
    "mcpServers": {
        "label-lab": {
            "type": "stdio",
            "command": sys.executable,
            "args": ["-u", str(server)],
        }
    }
}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
print("Created", configuration)

.\.venv\Scripts\python.exe configure_client.py
Get-Content -LiteralPath .mcp.json