让专长 Agent 通过 MCP 读取工单并调用受控工具：资源、权限边界与失败恢复

uv add "mcp==2.3.0"

import json

from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ResourceError

server = MCPServer("ticket-bridge")

# 代替你的工单系统 API。除了 add_internal_note，
# 服务器不会改动 TICKETS。
TICKETS = {
    "481": {"subject": "Checkout fails on Safari", "severity": 2,
            "report": "Customer reports the pay button does nothing on Safari 17.",
            "notes": []},
}


@server.resource("tickets://{ticket_id}", name="ticket",
                 description="One support ticket: subject, severity, report, notes")
def ticket(ticket_id: str) -> str:
    record = TICKETS.get(ticket_id)
    if record is None:
        # 可预期的失败：客户端原样看到这条消息。
        raise ResourceError(f"no ticket {ticket_id}")
    return json.dumps(record, ensure_ascii=False)


@server.tool()
def add_internal_note(ticket_id: str, note: str) -> str:
    """Append an internal note to a ticket. Staff-visible only.

    Args:
        ticket_id: Ticket identifier, e.g. "481".
        note: One or two sentences the next reader can act on.
    """
    record = TICKETS.get(ticket_id)
    if record is None:
        raise ResourceError(f"no ticket {ticket_id}")
    if len(note) > 500:
        raise ResourceError("note too long; keep it under 500 characters")
    record["notes"].append(note)
    return f"note added to {ticket_id}; {len(record['notes'])} total"


if __name__ == "__main__":
    server.run("stdio")

import asyncio
import os

from mcp import Client, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp_types import TextContent


async def main() -> None:
    params = StdioServerParameters(
        command="python",
        args=["ticket_server.py"],
        # 从环境注入只读员工 token，不写进代码。
        env={"TICKET_API_TOKEN": os.environ["TICKET_API_TOKEN"]},
    )
    async with Client(stdio_client(params)) as client:
        tools = await client.list_tools()
        print([t.name for t in tools.tools])  # ['add_internal_note']

        result = await client.call_tool(
            "add_internal_note",
            {"ticket_id": "481", "note": "Reproduced on Safari 17; console shows a TypeError."},
        )
        texts = [b.text for b in result.content if isinstance(b, TextContent)]
        print(result.is_error, texts)  # False ['note added to 481; 1 total']


asyncio.run(main())