Running the project

from claude_agent_sdk import (AgentDefinition, ClaudeAgentOptions, HookMatcher,
                              PermissionResultAllow, PermissionResultDeny)
from shop_tools import shop

APPROVED, LOG = {"A17"}, []


async def check(tool_name, input_data, context):
    # only a refund reaches here, because the lookup is pre-approved
    if input_data["order_id"] in APPROVED:
        return PermissionResultAllow()
    return PermissionResultDeny(message="Refunds need a person's approval first.")


async def audit(input_data, tool_use_id, context):
    # hooks run before the permission decision, so a denied call is logged too
    LOG.append(f"{input_data['tool_name']} {input_data['tool_input']['order_id']}")
    return {}


shipping = AgentDefinition(description="Answers questions about where an order is.",
                           prompt="You answer shipping questions for a shop.",
                           tools=["mcp__shop__lookup_order"])
options = ClaudeAgentOptions(
    mcp_servers={"shop": shop}, can_use_tool=check, max_turns=4,
    allowed_tools=["mcp__shop__lookup_order"], system_prompt="You run a shop's help desk.",
    agents={"shipping": shipping},
    hooks={"PreToolUse": [HookMatcher(hooks=[audit])]})
QUESTIONS = ["Where are my orders B22 and A17?", "When did it ship?",
             "Ask shipping about A17", "Please refund my order A17",
             "Please refund my order C40"]

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

from shop_desk import ShopDesk


def describe(message):
    # the two messages with no content blocks are read from their own fields
    if isinstance(message, SystemMessage):
        return f"system     {message.subtype}"
    if isinstance(message, ResultMessage):
        return f"result     {message.subtype}: {message.result or message.errors}"

    # a type this course does not use prints its name rather than failing
    if not isinstance(message, (AssistantMessage, UserMessage)):
        return type(message).__name__
    return describe_block(message.content[0])


def describe_block(block):
    if isinstance(block, TextBlock):
        return f"assistant  {block.text}"
    if isinstance(block, ToolUseBlock):
        return f"assistant  calls {block.name} {block.input}"

    # a tool result sits on the user's side, and its error flag is worth seeing
    if isinstance(block, ToolResultBlock):
        flag = " (error)" if block.is_error else ""
        return f"user       result{flag}: {block.content}"
    return type(block).__name__


async def watch(prompt, options, model=None):
    async for message in query(prompt=prompt, options=options,
                               transport=ShopDesk(options, model)):
        print(describe(message))

import asyncio
import itertools
import json
import re

from claude_agent_sdk import ClaudeAgentOptions, Transport

ORDER = re.compile(r"\b[A-Z]\d+\b")
SESSIONS = itertools.count(1)


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

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

    def reply(self, said, results, tools=(), agents=()):
        # a scripted reply lets a lesson force a step the stand-in would not pick
        if self.script:
            return self.script.pop(0)
        if results:
            return {"text": " ".join(results)}

        # naming an agent in the message is this stand-in's way of delegating
        for name in agents:
            if name in said[-1]:
                return {"calls": [("Agent", {"subagent_type": name, "prompt": said[-1]})]}
        return self.pick(said, tools)

    def pick(self, said, tools):
        # the newest message first, then the last id named earlier in the chat
        orders = ORDER.findall(said[-1]) or ORDER.findall(" ".join(said))[-1:]
        wanted = "refund_order" if "refund" in said[-1].lower() else "lookup_order"
        tool = next((name for name in tools if name.endswith(wanted)), None)
        if orders and tool:
            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?"}


class ShopDesk(Transport):
    """Stands in for the Claude Code binary: it runs the loop over JSON lines."""

    def __init__(self, options=None, model=None):
        self.options = options or ClaudeAgentOptions()
        self.model = model or ShopModel()

        # a counted id, not a UUID, so the printed output is the same every run
        self.session_id = f"session-{next(SESSIONS)}"
        self.inbox, self.outbox = asyncio.Queue(), asyncio.Queue()
        self.prompts, self.said = asyncio.Queue(), []
        self.ids, self.setup = itertools.count(1), {}
        self.waiting, self.tools = {}, {}
        self.ready = False

    async def connect(self):
        self.ready = True

    def is_ready(self):
        return self.ready

    async def write(self, data):
        # one JSON line at a time, which is what the SDK writes to a real process
        await self.inbox.put(json.loads(data))

    async def end_input(self):
        # for a real process this would close its standard input
        await self.inbox.put(None)

    async def close(self):
        self.ready = False

    async def send(self, message):
        await self.outbox.put(message)

    async def read_messages(self):
        # the listener and the worker run alongside the SDK's own read loop
        asyncio.create_task(self.listen())
        self.worker = asyncio.create_task(self.work())

        while (message := await self.outbox.get()) is not None:
            if isinstance(message, Exception):
                raise message
            yield message

    async def listen(self):
        while (message := await self.inbox.get()) is not None:
            # a control request is the SDK asking a question; initialize is first
            if message["type"] == "control_request":
                if message["request"]["subtype"] == "initialize":
                    self.setup = message["request"]
                await self.respond(message["request_id"])

            # a control response answers a question the desk itself asked
            elif message["type"] == "control_response":
                answer = message["response"]
                self.waiting.pop(answer["request_id"]).set_result(answer)
            elif message["type"] == "user":
                await self.prompts.put(message["message"]["content"])

        # the input has ended: finish the work in hand, then close the stream
        await self.prompts.put(None)
        await self.worker
        await self.send(None)

    async def respond(self, request_id):
        # the stand-in says yes to every request the SDK makes of it
        await self.send({"type": "control_response", "response": {
            "subtype": "success", "request_id": request_id, "response": {}}})

    async def ask(self, request):
        # a future per request, so the answer can be awaited where it was asked
        request_id = f"desk_{next(self.ids)}"
        self.waiting[request_id] = asyncio.get_running_loop().create_future()
        await self.send({"type": "control_request", "request_id": request_id,
                         "request": request})
        return await self.waiting[request_id]

    async def work(self):
        while (prompt := await self.prompts.get()) is not None:
            await self.turn(prompt)

    async def turn(self, prompt):
        self.said.append(prompt)
        if not self.tools:
            self.tools = await self.find_tools()
        await self.send({"type": "system", "subtype": "init", "model": "shop-desk",
                         "session_id": self.session_id, "tools": list(self.tools)})

        # the agent definitions arrived in the initialize request, not in the options
        agents = self.setup.get("agents") or {}
        results, turns, limit = [], 0, self.options.max_turns
        while "calls" in (step := self.model.reply(self.said, results, self.tools, agents)):
            turns += 1
            if limit and turns > limit:
                return await self.finish("error_max_turns", turns,
                                         f"Reached maximum number of turns ({limit})")
            results = [await self.use(name, args) for name, args in step["calls"]]

        await self.say(step["text"])
        await self.finish("success", turns + 1, step["text"])

    async def mcp(self, server, method, params):
        message = {"jsonrpc": "2.0", "id": next(self.ids), "method": method,
                   "params": params}
        answer = await self.ask({"subtype": "mcp_message", "server_name": server,
                                 "message": message})
        return answer["response"]["mcp_response"]["result"]

    async def find_tools(self):
        tools = {}
        for server in self.options.mcp_servers:
            await self.mcp(server, "initialize", {"protocolVersion": "2025-06-18",
                           "capabilities": {}, "clientInfo": {"name": "desk", "version": "1"}})

            # the full name carries the server, which is how a rule matches one tool
            for tool in (await self.mcp(server, "tools/list", {}))["tools"]:
                tools[f"mcp__{server}__{tool['name']}"] = (server, tool["name"])
        return tools

    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))

        # Agent is the tool that starts a subagent, so it skips permissions and hooks
        if name == "Agent":
            text, failed = await self.subagent(args, call), False
        else:
            allowed, args = await self.permit(name, args, call)
            text, failed = (await self.run(name, args)) 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

    async def run(self, name, args):
        server, tool = self.tools[name]
        result = await self.mcp(server, "tools/call", {"name": tool, "arguments": args})
        return result["content"][0]["text"], result.get("isError", False)

    async def permit(self, name, args, call):
        # hooks are the first step, so nothing below here can skip them
        verdict, args = await self.pre_hooks(name, args, call)
        if verdict in ("allow", "deny"):
            return verdict == "allow", args

        o = self.options
        if o.permission_mode == "bypassPermissions" or name in o.allowed_tools:
            return True, args

        # nothing approved the call and there is nobody left to ask
        if o.can_use_tool is None:
            return False, f"Permission to use {name} was not granted."
        return await self.ask_permission(name, args, call)

    async def ask_permission(self, name, args, call):
        # the SDK runs your callback and writes its answer back on this channel
        decision = (await self.ask({"subtype": "can_use_tool", "tool_name": name,
                                    "input": args, "tool_use_id": call,
                                    "permission_suggestions": None,
                                    "blocked_path": None}))["response"]

        # an allow may hand back different arguments to run with
        if decision["behavior"] == "allow":
            return True, decision["updatedInput"]
        return False, decision["message"]

    async def pre_hooks(self, name, args, call):
        verdict, data = None, {"tool_name": name, "tool_input": args, "tool_use_id": call}
        for group in (self.setup.get("hooks") or {}).get("PreToolUse", []):
            # a matcher of None means the group runs for every tool
            if group["matcher"] and not re.fullmatch(group["matcher"], name):
                continue

            # the SDK sent ids, not functions, so the call goes back by id
            for callback in group["hookCallbackIds"]:
                answer = await self.ask({
                    "subtype": "hook_callback", "callback_id": callback,
                    "tool_use_id": call,
                    "input": {"hook_event_name": "PreToolUse", "cwd": ".",
                              "session_id": self.session_id, **data}})
                decision = answer["response"].get("hookSpecificOutput", {})

                # one deny ends it, and the other hooks' opinions no longer matter
                if decision.get("permissionDecision") == "deny":
                    return "deny", decision.get("permissionDecisionReason", "Blocked")
                verdict = decision.get("permissionDecision", verdict)
                args = decision.get("updatedInput", args)
        return verdict, args

    async def subagent(self, args, call):
        spec = self.setup["agents"][args["subagent_type"]]

        # a tool left out of the definition is hidden from the subagent
        allowed = spec.get("tools") or list(self.tools)
        tools = {name: where for name, where in self.tools.items() if name in allowed}

        # a fresh model with only this prompt, so the main conversation stays clean
        said, results = [args["prompt"]], []
        while "calls" in (step := ShopModel().reply(said, results, tools)):
            results = [await self.use(name, a, call) for name, a in step["calls"]]
        await self.say(step["text"], call)
        return step["text"]

    async def say(self, text, parent=None):
        # only the main session streams; a subagent's text arrives whole
        if self.options.include_partial_messages and parent is None:
            for word in text.split(" "):
                delta = {"type": "text_delta", "text": word + " "}
                await self.send({"type": "stream_event", "uuid": f"ev_{next(self.ids)}",
                                 "session_id": self.session_id, "event": {
                                     "type": "content_block_delta", "index": 0,
                                     "delta": delta}})

        # the whole message still follows, so code written earlier keeps working
        await self.send(self.message("assistant", [{"type": "text", "text": text}], parent))

    async def finish(self, subtype, turns, text):
        ok = subtype == "success"
        result = {"type": "result", "subtype": subtype, "is_error": not ok,
                  "duration_ms": 0, "duration_api_ms": 0, "num_turns": turns,
                  "session_id": self.session_id, "total_cost_usd": 0.0,
                  "usage": {"input_tokens": len(" ".join(self.said).split()),
                            "output_tokens": len(text.split())}}

        # cost is zero and usage counts words, because no model was called
        result.update({"result": text} if ok else {"errors": [text]})
        await self.send(result)

    def message(self, kind, content, parent=None):
        body = {"role": kind, "content": content}
        if kind == "assistant":
            body.update(model="shop-desk", id=f"msg_{next(self.ids)}")
        return {"type": kind, "message": body, "parent_tool_use_id": parent,
                "session_id": self.session_id}

from claude_agent_sdk import create_sdk_mcp_server, tool

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


# 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}]}

REFUNDED = []


@tool("refund_order", "Refund an order by its id. The money goes back to the customer.",
      {"order_id": str})
async def refund_order(args):
    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])