Project: MCP support desk

from typing import Annotated, Literal

from pydantic import BaseModel

from mcp.server import MCPServer
from mcp.server.mcpserver import AcceptedElicitation, Elicit, ElicitationResult, Resolve
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import ToolAnnotations

mcp = MCPServer("Shop support", log_level="WARNING")

ORDERS = {
    "A17": {"item": "blue mug", "status": "shipped", "total": 12.50},
    "B42": {"item": "desk lamp", "status": "delivered", "total": 48.00},
}

class Order(BaseModel):
    id: str
    item: str
    status: Literal["processing", "shipped", "delivered"]
    total: float


class Confirm(BaseModel):
    approve: bool

@mcp.tool(annotations=ToolAnnotations(read_only_hint=True))
def lookup_order(order_id: str) -> Order:
    """Look up an order by its id."""
    if order_id not in ORDERS:
        raise ToolError(f"No order with id {order_id!r}. Order ids look like A17.")
    return Order(id=order_id, **ORDERS[order_id])

async def approve_refund(order_id: str) -> Confirm | Elicit[Confirm]:
    """Refunds of 20 or less go through; larger ones need a person."""
    if order_id in ORDERS and ORDERS[order_id]["total"] <= 20:
        return Confirm(approve=True)
    return Elicit(f"Approve a refund for order {order_id}?", Confirm)


@mcp.tool(title="Refund an order", annotations=ToolAnnotations(read_only_hint=False, idempotent_hint=False))
async def refund_order(
    order_id: str,
    reason: str,
    confirm: Annotated[ElicitationResult[Confirm], Resolve(approve_refund)],
) -> str:
    """Refund the full total of an order."""
    if order_id not in ORDERS:
        raise ToolError(f"No order with id {order_id!r}.")
    if not (isinstance(confirm, AcceptedElicitation) and confirm.data.approve):
        return f"The refund for {order_id} was not approved, so nothing was refunded."
    return f"Refunded {ORDERS[order_id]['total']:.2f} for order {order_id}."

@mcp.resource("policy://refunds")
def refund_policy() -> str:
    """The shop's refund policy."""
    return "Full refund within 30 days of delivery."


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

import asyncio
import sys

from mcp import Client
from mcp.types import ElicitResult

from agent import run_agent

APPROVE = sys.argv[1] == "approve"


async def supervisor(context, params):
    print(f"  supervisor asked: {params.message} -> {'yes' if APPROVE else 'no'}")
    return ElicitResult(action="accept", content={"approve": APPROVE})

async def main():
    messages = [
        "Where is my order A17?",
        "Where is my order Z9?",
        "Please refund order A17, the handle broke",
        "Please refund order B42, it arrived broken",
    ]
    async with Client("http://127.0.0.1:8200/mcp", elicitation_callback=supervisor) as client:
        for message in messages:
            print(message)
            print(" ", await run_agent(client, message))


asyncio.run(main())

python shop.py &
sleep 3
python desk.py decline
kill $!

Where is my order A17?
  calling lookup_order {'order_id': 'A17'}
  Your order A17 (blue mug) has been shipped.
Where is my order Z9?
  calling lookup_order {'order_id': 'Z9'}
  I’m sorry, but I can’t find an order with the ID “Z9.” Could you double‑check the order number and let me know the correct one?
Please refund order A17, the handle broke
  calling refund_order {'order_id': 'A17', 'reason': 'the handle broke'}
  Order A17 has been refunded $12.50.
Please refund order B42, it arrived broken
  calling refund_order {'order_id': 'B42', 'reason': 'arrived broken'}
  supervisor asked: Approve a refund for order B42? -> no
  The refund for order B42 could not be approved.

python shop.py &
sleep 3
python desk.py approve
kill $!

Where is my order A17?
  calling lookup_order {'order_id': 'A17'}
  Your order A17 (blue mug) has been shipped.
Where is my order Z9?
  calling lookup_order {'order_id': 'Z9'}
  I’m sorry, but I can’t find an order with the ID “Z9.” Could you double‑check the order number and let me know the correct one?
Please refund order A17, the handle broke
  calling refund_order {'order_id': 'A17', 'reason': 'the handle broke'}
  Order A17 has been refunded $12.50.
Please refund order B42, it arrived broken
  calling refund_order {'order_id': 'B42', 'reason': 'arrived broken'}
  supervisor asked: Approve a refund for order B42? -> yes
  Your order B42 has been refunded in full.

import asyncio

from mcp import Client, MCPError
from shop import mcp


async def main():
    async with Client(mcp) as client:
        try:
            await client.call_tool("refund_order", {"order_id": "B42", "reason": "broken"})
        except MCPError as error:
            print("refund blocked:", error)


asyncio.run(main())

refund blocked: Client did not declare the form elicitation capability required by resolver 'shop:approve_refund'

import pytest
from mcp import Client, MCPError
from mcp.types import ElicitResult

from shop import mcp


@pytest.fixture
def anyio_backend():
    return "asyncio"


async def decline(context, params):
    return ElicitResult(action="decline")


@pytest.mark.anyio
async def test_small_refund_needs_no_approval():
    async with Client(mcp) as client:
        result = await client.call_tool("refund_order", {"order_id": "A17", "reason": "broken"})
    assert result.content[0].text == "Refunded 12.50 for order A17."


@pytest.mark.anyio
async def test_declined_refund_refunds_nothing():
    async with Client(mcp, elicitation_callback=decline) as client:
        result = await client.call_tool("refund_order", {"order_id": "B42", "reason": "broken"})
    assert "not approved" in result.content[0].text


@pytest.mark.anyio
async def test_large_refund_fails_without_a_way_to_ask():
    async with Client(mcp) as client:
        with pytest.raises(MCPError):
            await client.call_tool("refund_order", {"order_id": "B42", "reason": "broken"})

pytest -q

...  [100%]
3 passed in 0.35s