Permission modes

from claude_agent_sdk import create_sdk_mcp_server, tool

ORDERS = {"A17": "Order A17 shipped on 3 March."}
REFUNDED = []


# the description is what Claude reads to decide this is the right tool
@tool("lookup_order", "Look up an order's shipping status by its id, such as A17.",
      {"order_id": str})
async def lookup_order(args):
    order = args["order_id"]

    # an unknown id is answered in words, not raised, so Claude can pass it on
    status = ORDERS.get(order, f"No order {order} was found.")
    return {"content": [{"type": "text", "text": status}]}


# the money moves, so this is the call that must not run unasked
@tool("refund_order", "Refund an order by its id. The money goes back to the customer.",
      {"order_id": str})
async def refund_order(args):
    # recorded so a lesson can show whether the refund happened at all
    REFUNDED.append(args["order_id"])
    text = f"Order {args['order_id']} has been refunded."
    return {"content": [{"type": "text", "text": text}]}


shop = create_sdk_mcp_server("shop", tools=[lookup_order, refund_order])

import asyncio
import itertools
import json
import os
import re

from claude_agent_sdk import (AssistantMessage, ClaudeAgentOptions, ResultMessage,
                              SystemMessage, TextBlock, ToolResultBlock, ToolUseBlock,
                              Transport, UserMessage, query)

ORDER = re.compile(r"\b[A-Z]\d+\b")
SESSIONS = itertools.count(1)
WANTS = {"refund": "refund_order", "stock": "stock_level"}


def read_file(args):
    """What the desk serves for a Read call: the file's lines, numbered."""
    lines = open(args["file_path"]).read().splitlines()
    return "\n".join(f"{number:>5}  {line}" for number, line in enumerate(lines, 1))


BUILT_IN = {"Read": read_file}


class ShopModel:
    """Stands in for Claude: reads the conversation, picks the next step."""

    def __init__(self, script=None):
        self.script = list(script or [])
        self.searched = False

    def reply(self, said, results, tools):
        if self.script:
            return self.script.pop(0)

        # a search result is not an answer, so a searching turn does not stop here
        if results and not self.searched:
            return {"text": " ".join(results)}
        orders = ORDER.findall(said[-1]) or ORDER.findall(" ".join(said))[-1:]
        wanted = next((name for word, name in WANTS.items() if word in said[-1].lower()),
                      "lookup_order")
        tool = next((name for name in tools if name.endswith(wanted)), None)

        # the tool this message needs has no definition in context: search for it first
        if orders and tool is None and not self.searched and "ToolSearch" in tools:
            self.searched = True
            return {"calls": [("ToolSearch", {"query": wanted.replace("_", " ")})]}

        # the definition is in context now, so the call itself can be made
        if orders and tool:
            self.searched = False
            return {"calls": [(tool, {"order_id": order}) for order in orders]}
        return {"text": f"I cannot help with order {orders[0]} yet." if orders
                else "Which order is this about?"}

    async def permit(self, name, args, call):
        # the mode step approves what reaches it, or sends the call on
        if self.options.permission_mode == "bypassPermissions":
            return True, args
        return False, f"Permission to use {name} was not granted."

    async def use(self, name, args, parent=None):
        call = f"toolu_{next(self.ids)}"
        block = {"type": "tool_use", "id": call, "name": name, "input": args}
        await self.send(self.message("assistant", [block], parent))

        # a refused call reuses args as the text to send back, so Claude is told why
        allowed, args = await self.permit(name, args, call)
        text, failed = (await self.run(name, args, call)) if allowed else (args, True)

        result = {"type": "tool_result", "tool_use_id": call, "content": text,
                  "is_error": failed}
        await self.send(self.message("user", [result], parent))
        return text