wehub-resource-sync
49b9bb6724
Deploy Docs / deploy-docs (push) Failing after 1s
Conformance Tests / client-conformance (push) Failing after 3s
Conformance Tests / server-conformance (push) Failing after 1s
GitHub Actions Security Analysis / zizmor (push) Failing after 1s
CI / checks (push) Failing after 59m20s
CI / all-green (push) Waiting to run
128 lines
3.8 KiB
Python
128 lines
3.8 KiB
Python
"""Resolver DI: the refund amount is computed by resolvers from the order record — `cents` never appears in the
|
|
tool's input schema, so the model cannot supply or inflate it."""
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Annotated
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from mcp.server.mcpserver import (
|
|
AcceptedElicitation,
|
|
Elicit,
|
|
ElicitationResult,
|
|
MCPServer,
|
|
Resolve,
|
|
)
|
|
from mcp.server.mcpserver.exceptions import ToolError
|
|
from stories._hosting import run_server_from_args
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Line:
|
|
sku: str
|
|
cents: int
|
|
physical: bool
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Order:
|
|
order_id: str
|
|
lines: tuple[Line, ...]
|
|
|
|
|
|
ORDERS: dict[str, Order] = {
|
|
"ORD-7001": Order("ORD-7001", (Line("ebook-fieldnotes", 1500, physical=False),)),
|
|
"ORD-7002": Order(
|
|
"ORD-7002",
|
|
(
|
|
Line("enamel-mug", 1800, physical=True),
|
|
Line("canvas-tote", 2400, physical=True),
|
|
Line("sticker-pack", 600, physical=False),
|
|
),
|
|
),
|
|
}
|
|
|
|
|
|
class Scope(BaseModel):
|
|
"""Which items to refund: the whole order, or a single SKU."""
|
|
|
|
full: bool
|
|
sku: str = ""
|
|
|
|
|
|
class RestockChoice(BaseModel):
|
|
restock: bool
|
|
|
|
|
|
class Receipt(BaseModel):
|
|
order_id: str
|
|
refunded_cents: int
|
|
restocked: bool
|
|
reason: str
|
|
|
|
|
|
def load_order(order_id: str) -> Order:
|
|
order = ORDERS.get(order_id)
|
|
if order is None:
|
|
raise ToolError(f"unknown order {order_id!r}")
|
|
return order
|
|
|
|
|
|
def refund_scope(order_id: str, order: Annotated[Order, Resolve(load_order)]) -> Scope | Elicit[Scope]:
|
|
if len(order.lines) == 1:
|
|
return Scope(full=True)
|
|
skus = ", ".join(line.sku for line in order.lines)
|
|
return Elicit(f"{order_id} has several items ({skus}). Refund the whole order, or one SKU?", Scope)
|
|
|
|
|
|
def _scoped(order: Order, scope: Scope) -> tuple[Line, ...]:
|
|
"""The lines a scope covers. The SKU was typed by a human — validate it against the order."""
|
|
if scope.full:
|
|
return order.lines
|
|
lines = tuple(line for line in order.lines if line.sku == scope.sku)
|
|
if not lines:
|
|
raise ToolError(f"order has no item {scope.sku!r}")
|
|
return lines
|
|
|
|
|
|
def refund_amount(
|
|
order: Annotated[Order, Resolve(load_order)],
|
|
scope: Annotated[Scope, Resolve(refund_scope)],
|
|
) -> int:
|
|
return sum(line.cents for line in _scoped(order, scope))
|
|
|
|
|
|
def ask_restock(
|
|
order: Annotated[Order, Resolve(load_order)],
|
|
scope: Annotated[Scope, Resolve(refund_scope)],
|
|
) -> RestockChoice | Elicit[RestockChoice]:
|
|
physical = [line.sku for line in _scoped(order, scope) if line.physical]
|
|
if not physical:
|
|
return RestockChoice(restock=False)
|
|
return Elicit(f"The refund includes physical items ({', '.join(physical)}). Return them to stock?", RestockChoice)
|
|
|
|
|
|
def build_server() -> MCPServer:
|
|
# Elicited answers ride between rounds in a requestState the SDK seals by default;
|
|
# see mrtr/ for the full security walk-through.
|
|
mcp = MCPServer("refund-desk")
|
|
|
|
@mcp.tool(description="Refund an order. The amount comes from the order record, not from the caller.")
|
|
def refund_order(
|
|
order_id: str,
|
|
reason: str,
|
|
cents: Annotated[int, Resolve(refund_amount)],
|
|
restock: Annotated[ElicitationResult[RestockChoice], Resolve(ask_restock)],
|
|
) -> Receipt:
|
|
# `restock` keeps the full elicitation outcome: a declined restock still refunds. A plain
|
|
# (non-Elicit) resolver return arrives wrapped as an accepted outcome, so the fast path
|
|
# lands in the same `AcceptedElicitation` branch.
|
|
restocked = isinstance(restock, AcceptedElicitation) and restock.data.restock
|
|
return Receipt(order_id=order_id, refunded_cents=cents, restocked=restocked, reason=reason)
|
|
|
|
return mcp
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_server_from_args(build_server)
|