Project: a support desk agent over MCP

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 %1

python shop.py &
sleep 3
python desk.py approve
kill %1

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