MCP 工具详解：定义、结果、错误与跨版本差异

"""Official Python SDK (mcp==2.3.0) server with one tool per result shape, http://127.0.0.1:8200/mcp, no auth."""
from typing import Annotated

from mcp.server.mcpserver import MCPServer
from pydantic import BaseModel, Field

server = MCPServer("tools-ref")


class Weather(BaseModel):
    temperature: float
    conditions: str


class User(BaseModel):
    id: str
    name: str


@server.tool()
def get_weather(location: Annotated[str, Field(min_length=2)]) -> Weather:
    """Return fixed weather for a location."""
    return Weather(temperature=22.5, conditions="Partly cloudy")


@server.tool()
def list_users() -> list[User]:
    """Return two users as an array."""
    return [User(id="1", name="Alice"), User(id="2", name="Bob")]


@server.tool()
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b


@server.tool()
def get_current_time() -> str:
    """Take no arguments."""
    return "2026-10-26T00:00:00Z"


@server.tool()
def explode() -> str:
    """Always raise."""
    raise RuntimeError("boom")


if __name__ == "__main__":
    server.run("streamable-http", port=8200)

list             HTTP 200 result   ttlMs=0 cacheScope=private resultType=complete tools=5 | ListToolsResultResponse valid
weather-ok       HTTP 200 result   1 text blocks, structuredContent {"temperature":22.5,"conditions":"Partly cloudy"} | CallToolResultResponse valid
sum-ok           HTTP 200 result   1 text blocks, structuredContent {"result":3} | CallToolResultResponse valid
users-ok         HTTP 200 result   2 text blocks, structuredContent {"result":[{"id":"1","name":"Alice"},{"id":"2","name":"Bob"}]} | CallToolResultResponse valid
time-no-args     HTTP 200 result   1 text blocks, structuredContent {"result":"2026-10-26T00:00:00Z"} | CallToolResultResponse valid
bad-type         HTTP 200 isError  text 'Error executing tool add: 1 validation error for addArguments\n' link=errors.pydantic.dev | CallToolResultResponse valid
bad-constraint   HTTP 200 isError  text 'Error executing tool get_weather: 1 validation error for get_w' link=errors.pydantic.dev | CallToolResultResponse valid
missing-argument HTTP 200 isError  text 'Error executing tool get_weather: 1 validation error for get_w' link=errors.pydantic.dev | CallToolResultResponse valid
handler-raises   HTTP 200 isError  text 'Error executing tool explode' | CallToolResultResponse valid
unknown-tool     HTTP 200 isError  text 'Unknown tool: nope' | CallToolResultResponse valid
missing-name     HTTP 400 error    code -32602 'Invalid request parameters' | JSONRPCErrorResponse valid
arguments-array  HTTP 400 error    code -32602 'Invalid request parameters' | JSONRPCErrorResponse valid

#!/usr/bin/env python3
"""Lint the tool definitions of an MCP server against spec 2026-07-28, and optionally call one tool.

Usage: lint_tools.py SOURCE [--call NAME [--args JSON]] [--token T]
SOURCE: URL of a Streamable HTTP endpoint, or a JSON file ("-" = stdin) holding a tools/list response,
its result, or a tools array. The list result itself is checked too (resultType, ttlMs, cacheScope) unless the
file holds a bare tools array. Schema checks need `pip install jsonschema` (skipped with a WARN without it).
Levels: FAIL = a MUST of the spec is violated, WARN = a SHOULD or a recommendation is not followed.
Exit code: 0 no FAIL, 1 at least one FAIL, 2 the source cannot be read.
"""
import argparse
import http.client
import json
import re
import sys
from urllib.parse import urlsplit

try:
    import jsonschema
except ImportError:
    jsonschema = None

VERSION = "2026-07-28"
NAME_RE = re.compile(r"^[A-Za-z0-9_.-]{1,128}$")
TCHAR_RE = re.compile(r"^[!#$%&'*+\-.^_`|~0-9A-Za-z]+$")
HEADER_TYPES = {"integer", "string", "boolean"}
counts = {"PASS": 0, "WARN": 0, "FAIL": 0}


def report(name, ok, detail, level="FAIL"):
    status = "PASS" if ok else level
    counts[status] += 1
    print("%s %s: %s" % (status, name, detail))


def rpc(url, method, params=None, token=None):
    """POST one 2026-07-28 request and return (HTTP status, JSON-RPC message or None)."""
    parts = urlsplit(url)
    cls = http.client.HTTPSConnection if parts.scheme == "https" else http.client.HTTPConnection
    meta = {"io.modelcontextprotocol/protocolVersion": VERSION,
            "io.modelcontextprotocol/clientInfo": {"name": "lint-tools", "version": "0"},
            "io.modelcontextprotocol/clientCapabilities": {}}
    headers = {"Accept": "application/json, text/event-stream", "Content-Type": "application/json",
               "MCP-Protocol-Version": VERSION, "Mcp-Method": method}
    if params and "name" in params:
        headers["Mcp-Name"] = params["name"]
    if token:
        headers["Authorization"] = "Bearer " + token
    body = json.dumps({"jsonrpc": "2.0", "id": 1, "method": method, "params": {**(params or {}), "_meta": meta}})
    conn = cls(parts.netloc, timeout=15)
    conn.request("POST", parts.path or "/", body, headers)
    resp = conn.getresponse()
    raw = resp.read().decode("utf-8", "replace")
    conn.close()
    try:
        return resp.status, json.loads(raw)
    except ValueError:
        for line in reversed(raw.splitlines()):  # SSE: the last data line holding a response
            if line.startswith("data:"):
                try:
                    return resp.status, json.loads(line[5:].strip())
                except ValueError:
                    continue
    return resp.status, None


def load_tools(source, token):
    """Return (tools, envelope): envelope is the first tools/list result, None for a bare tools array."""
    if source.startswith(("http://", "https://")):
        tools, cursor, envelope = [], None, None
        for _ in range(50):
            status, msg = rpc(source, "tools/list", {"cursor": cursor} if cursor else None, token)
            result = (msg or {}).get("result")
            if status != 200 or not isinstance(result, dict) or not isinstance(result.get("tools"), list):
                raise ValueError("tools/list failed: HTTP %s, %s" % (status, json.dumps(msg)[:200]))
            tools += result["tools"]
            envelope = envelope or result
            cursor = result.get("nextCursor")
            if not cursor:
                return tools, envelope
        raise ValueError("tools/list still paginating after 50 pages")
    data = json.load(sys.stdin if source == "-" else open(source, encoding="utf-8"))
    data = data.get("result", data) if isinstance(data, dict) else data
    tools = data.get("tools") if isinstance(data, dict) else data
    if not isinstance(tools, list):
        raise ValueError("no tools array found in %s" % source)
    return tools, data if isinstance(data, dict) else None


def dialect(schema):
    uri = schema.get("$schema", "") if isinstance(schema, dict) else ""
    return jsonschema.Draft7Validator if "draft-07" in uri else jsonschema.Draft202012Validator


def schema_problem(schema):
    try:
        dialect(schema).check_schema(schema)
    except jsonschema.SchemaError as exc:
        return exc.message[:100]
    return None


def header_annotations(schema):
    """Yield (name, value, chain, reachable) for every x-mcp-header in a schema; reachable means a pure `properties` chain."""
    def walk(node, chain, reachable):
        if not isinstance(node, dict):
            return
        if "x-mcp-header" in node:
            yield node, chain, reachable
        for key, value in node.items():
            if key == "properties" and isinstance(value, dict):
                for prop, sub in value.items():
                    yield from walk(sub, chain + [prop], reachable)
            elif key not in ("x-mcp-header", "enum", "const", "default", "examples"):
                for sub in (value if isinstance(value, list) else
                            list(value.values()) if key in ("$defs", "definitions", "patternProperties") else [value]):
                    yield from walk(sub, chain + [key], False)
    yield from walk(schema, [], True)


def header_problems(schema):
    problems, seen = [], {}
    for node, chain, reachable in header_annotations(schema):
        value, where = node["x-mcp-header"], "/".join(chain) or "(root)"
        if not reachable:
            problems.append("%s: not statically reachable" % where)
        if not isinstance(value, str) or not TCHAR_RE.match(value):
            problems.append("%s: %r is not a non-empty HTTP token" % (where, value))
            continue
        if value.lower() in seen:
            problems.append("%s: duplicates %s (case-insensitive)" % (where, seen[value.lower()]))
        seen[value.lower()] = where
        kind = node.get("type")
        kinds = set(kind) if isinstance(kind, list) else {kind}
        if not kinds <= HEADER_TYPES:
            problems.append("%s: type %s is not integer, string or boolean" % (where, kind))
    return problems


def lint_envelope(result):
    """tools/list results of 2026-07-28 MUST carry resultType, and caching hints with ttlMs >= 0."""
    if result is None:
        return
    ttl, scope = result.get("ttlMs"), result.get("cacheScope")
    problems = []
    if not isinstance(result.get("resultType"), str):
        problems.append("resultType is missing")
    if not isinstance(ttl, int) or isinstance(ttl, bool) or ttl < 0:
        problems.append("ttlMs is %r, want an integer >= 0" % ttl)
    if scope not in ("public", "private"):
        problems.append("cacheScope is %r, want 'public' or 'private'" % scope)
    report("list-envelope", not problems, "; ".join(problems) if problems
           else "resultType=%s ttlMs=%d cacheScope=%s" % (result["resultType"], ttl, scope))


def lint(tools, envelope=None):
    lint_envelope(envelope)
    named = [t for t in tools if isinstance(t, dict) and isinstance(t.get("name"), str)]
    report("name-present", len(named) == len(tools), "%d of %d tools have a string name" % (len(named), len(tools)))
    bad = [t["name"] for t in named if not NAME_RE.match(t["name"])]
    report("name-charset", not bad, "bad names: %s" % bad if bad else "%d names match [A-Za-z0-9_.-]{1,128}" % len(named),
           "WARN")
    names = [t["name"] for t in named]
    dup = sorted({n for n in names if names.count(n) > 1})
    report("name-unique", not dup, "duplicates: %s" % dup if dup else "no duplicates", "WARN")
    root = [t["name"] for t in named if not (isinstance(t.get("inputSchema"), dict)
                                             and t["inputSchema"].get("type") == "object")]
    report("input-root-object", not root, "inputSchema is not an object schema with type=object: %s" % root
           if root else "every inputSchema is an object schema with type=object")
    open_schema = [t["name"] for t in named if isinstance(t.get("inputSchema"), dict)
                   and t["inputSchema"].get("type") == "object" and not t["inputSchema"].get("properties")
                   and t["inputSchema"].get("additionalProperties") is not False]
    report("no-params-closed", not open_schema,
           "no-parameter tools accept any object, add additionalProperties:false: %s" % open_schema
           if open_schema else "no-parameter tools are closed", "WARN")
    hdr = ["%s %s" % (t["name"], p) for t in named if isinstance(t.get("inputSchema"), dict)
           for p in header_problems(t["inputSchema"])]
    report("x-mcp-header", not hdr, "; ".join(hdr) if hdr else "annotations satisfy the constraints (or none used)")
    if jsonschema is None:
        report("schema-valid", False, "skipped, jsonschema is not installed", "WARN")
        return
    problems = ["%s.%s: %s" % (t["name"], k, p) for t in named for k in ("inputSchema", "outputSchema")
                if isinstance(t.get(k), dict) and (p := schema_problem(t[k]))]
    report("schema-valid", not problems, "; ".join(problems) if problems else "inputSchema / outputSchema are valid JSON Schema")


def call(url, tools, name, args, token):
    status, msg = rpc(url, "tools/call", {"name": name, "arguments": args}, token)
    if not isinstance(msg, dict):
        report("call", False, "HTTP %s without a JSON-RPC message" % status)
        return
    if "error" in msg:
        print("INFO call %s: protocol error, HTTP %s, code %s: %s" % (name, status, msg["error"].get("code"),
                                                                       msg["error"].get("message")))
        return
    result = msg.get("result") or {}
    if result.get("resultType") == "input_required":
        print("INFO call %s: input_required, retry with inputResponses (nothing more to check)" % name)
        return
    report("call-result-type", isinstance(result.get("resultType"), str), "resultType=%s" % result.get("resultType")
           if isinstance(result.get("resultType"), str) else "resultType is missing")
    if result.get("isError"):
        text = " ".join(c.get("text", "") for c in result.get("content", []) if c.get("type") == "text")
        print("INFO call %s: execution error (isError), HTTP %s: %s" % (name, status, text[:100].replace("\n", " ")))
        return
    print("INFO call %s: result, HTTP %s, %d content blocks, structuredContent %s" % (
        name, status, len(result.get("content", [])), "present" if "structuredContent" in result else "absent"))
    tool = next((t for t in tools if isinstance(t, dict) and t.get("name") == name), {})
    structured = result.get("structuredContent")
    if tool.get("outputSchema"):
        report("structured-present", "structuredContent" in result, "the tool declares an outputSchema")
        if jsonschema and "structuredContent" in result:
            errs = list(dialect(tool["outputSchema"])(tool["outputSchema"]).iter_errors(structured))
            report("structured-conforms", not errs, errs[0].message[:100] if errs else "structuredContent matches outputSchema")
    if "structuredContent" in result:
        copies = []
        for c in result.get("content", []):
            try:
                copies.append(c.get("type") == "text" and json.loads(c["text"]) == structured)
            except (ValueError, KeyError):
                copies.append(False)
        report("structured-text-copy", any(copies), "a text block holds the serialized structuredContent", "WARN")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("source")
    ap.add_argument("--call", help="name of a tool to call after the lint")
    ap.add_argument("--args", default="{}", help="JSON arguments for --call")
    ap.add_argument("--token", help="bearer token for the endpoint")
    args = ap.parse_args()
    if args.call and not args.source.startswith(("http://", "https://")):
        ap.error("--call needs a URL as SOURCE")
    try:
        tools, envelope = load_tools(args.source, args.token)
        arguments = json.loads(args.args)
    except (OSError, ValueError, http.client.HTTPException) as exc:
        print("ERROR %s" % exc)
        return 2
    print("INFO %d tools" % len(tools))
    lint(tools, envelope)
    if args.call:
        try:
            call(args.source, tools, args.call, arguments, args.token)
        except (OSError, http.client.HTTPException) as exc:
            print("ERROR %s" % exc)
            return 2
    print("%d pass, %d warn, %d fail" % (counts["PASS"], counts["WARN"], counts["FAIL"]))
    return 1 if counts["FAIL"] else 0


if __name__ == "__main__":
    sys.exit(main())