An in-process MCP server

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

    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 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:
            # MCP wants a handshake before a server will list anything
            await self.mcp(server, "initialize", {"protocolVersion": "2025-06-18",
                           "capabilities": {}, "clientInfo": {"name": "desk", "version": "1"}})

            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):
        # one id ties the request and its result together, as the message pair does
        call = f"toolu_{next(self.ids)}"
        block = {"type": "tool_use", "id": call, "name": name, "input": args}
        await self.send(self.message("assistant", [block], parent))

        text, failed = await self.run(name, args, call)
        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, call):
        # the full name carries the server it came from, so the call goes back there
        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 start(self, prompt):
        self.said.append(prompt)

        # the tool list is fetched once per desk, not once per question
        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)})

    async def turn(self, prompt):
        await self.start(prompt)

        # keep asking the model until it answers instead of calling a tool
        results, turns = [], 0
        while "calls" in (step := self.model.reply(self.said, results, self.tools)):
            turns += 1
            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"])

import asyncio
import itertools
import json
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)


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):
        # 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)}

        # 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.ready, self.tools = False, {}
        self.waiting = {}

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