Getting started with the Claude Agent SDK and mcp-windbg

Python

Clone agent-sdk-starter-python:

git clone https://github.com/svnscha/agent-sdk-starter-python.git
cd agent-sdk-starter-python
uv sync

Configure .env using the README, then run the greeting:

uv run --env-file .env agent.py

TypeScript

Clone agent-sdk-starter-typescript:

git clone https://github.com/svnscha/agent-sdk-starter-typescript.git
cd agent-sdk-starter-typescript
npm ci

Configure .env using the README, then run the greeting:

npm start

Python

uv add mcp==2.3.0 pydantic==2.14.1

TypeScript

npm install @modelcontextprotocol/[email protected] [email protected]

Python

Create triage.py:

import asyncio
import hashlib
import json
import ntpath
import os
import re
import signal
import sys
from pathlib import Path
from time import monotonic
from typing import Literal
from urllib.parse import quote

from claude_agent_sdk import create_sdk_mcp_server, tool
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from pydantic import BaseModel, ConfigDict, Field

from agent import log, run_agent


class Match(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)
    id: str = Field(pattern=r"^[a-f0-9]{16}$")
    reason: str = Field(min_length=1)


class Result(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)
    status: Literal["success", "failure"]
    summary: str = Field(min_length=1)
    report: str = Field(min_length=1)
    match: Match | None


class StoredResult(Result):
    estimatedCostUsd: float | None = Field(ge=0)


stop = asyncio.Event()


def stopping(signum, _frame):
    log(
        "worker.stopping",
        f"Received {signal.Signals(signum).name}; closing the active session.",
    )
    stop.set()


for sig in (signal.SIGINT, signal.SIGTERM):
    signal.signal(sig, stopping)

Path("dumps").mkdir(exist_ok=True)
Path("reports/triages").mkdir(parents=True, exist_ok=True)

TypeScript

Create triage.ts:

import { createHash } from "node:crypto";
import { mkdir, readdir, readFile, writeFile } from "node:fs/promises";
import path from "node:path";
import { setTimeout as sleep } from "node:timers/promises";
import {
  createSdkMcpServer,
  tool,
  type SDKResultMessage,
} from "@anthropic-ai/claude-agent-sdk";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import { z } from "zod";
import { log, runAgent } from "./agent.ts";

const Result = z
  .object({
    status: z.enum(["success", "failure"]),
    summary: z.string().min(1),
    report: z.string().min(1),
    match: z
      .object({
        id: z.string().regex(/^[a-f0-9]{16}$/),
        reason: z.string().min(1),
      })
      .strict()
      .nullable(),
  })
  .strict();

const StoredResult = Result.extend({
  estimatedCostUsd: z.number().nonnegative().nullable(),
});

const stop = new AbortController();
for (const signal of ["SIGINT", "SIGTERM"] as const) {
  process.once(signal, () => {
    log("worker.stopping", `Received ${signal}; closing the active session.`);
    stop.abort();
  });
}
process.once("exit", (code) => log("worker.stop", `exitCode=${code}`));
await mkdir("dumps", { recursive: true });
await mkdir("reports/triages", { recursive: true });

Python

async def investigate(name, on_result):
    url = os.environ.get("MCP_WINDBG_URL", "http://127.0.0.1:8000/mcp/")
    session_id = None
    opening = None
    command_lock = asyncio.Lock()
    listed = False
    inspected = False
    read_ids = set()

    def text(value):
        return {"content": [{"type": "text", "text": value}]}

    log("debugger.connect", "Connecting to MCP WinDbg.")
    async with streamable_http_client(url) as (read, write):
        async with ClientSession(read, write, read_timeout_seconds=190) as client:
            await asyncio.wait_for(client.initialize(), timeout=10)
            log("debugger.ready", "MCP WinDbg connected.")
            history = {
                file.stem: file.read_text(encoding="utf-8")
                for file in Path("reports/triages").glob("*.md")
                if re.fullmatch(r"[a-f0-9]{16}", file.stem)
            }

            # Serialize debugger calls, including cleanup after a failed call.
            async def call(tool_name, args, timeout=190):
                async with command_lock:
                    reply = await client.call_tool(
                        tool_name, args, read_timeout_seconds=timeout
                    )
                    output = "\n".join(
                        block.text for block in reply.content if block.type == "text"
                    )
                    if reply.is_error:
                        raise RuntimeError(output)
                    return output

            async def open_assigned_dump():
                nonlocal session_id
                output = await call(
                    "open_cdb_dump",
                    {
                        "dump_path": ntpath.join(os.environ["DUMP_HOST_DIR"], name),
                        "include_stack_trace": True,
                        "include_modules": True,
                        "timeout_seconds": 180,
                    },
                )
                match = re.search(
                    r"^session_id:\s*([a-z]+-[a-z0-9]+)", output, re.I | re.M
                )
                if not match:
                    raise RuntimeError("No debugger session ID.")
                session_id = match[1]
                return output

            @tool("open_dump", "Open the assigned dump.", {})
            async def open_dump(_args):
                nonlocal opening
                # Reuse the session if the agent asks to open the same dump again.
                if opening is None:
                    opening = asyncio.create_task(open_assigned_dump())
                return text(await opening)

            @tool(
                "run_command",
                "Inspect the open dump.",
                {
                    "type": "object",
                    "properties": {
                        "command": {
                            "type": "string",
                            "enum": ["!analyze -v", ".ecxr", "kv", "kb", "lm", "r"],
                        }
                    },
                    "required": ["command"],
                    "additionalProperties": False,
                },
            )
            async def run_command(args):
                nonlocal inspected
                if not session_id:
                    raise RuntimeError("Open the dump first.")
                output = await call(
                    "run_cdb_command",
                    {
                        "session_id": session_id,
                        "command": args["command"],
                    },
                )
                inspected = True
                return text(output)

            @tool("list_triages", "Find earlier triages.", {})
            async def list_triages(_args):
                nonlocal listed
                listed = True
                return text(
                    json.dumps(
                        [
                            {"id": triage_id, "summary": markdown.split("\n\n")[1]}
                            for triage_id, markdown in history.items()
                        ]
                    )
                )

            @tool("read_triage", "Read a possible match.", {"id": str})
            async def read_triage(args):
                triage_id = args["id"]
                if triage_id not in history:
                    raise RuntimeError("Unknown triage.")
                read_ids.add(triage_id)
                return text(history[triage_id])

            server = create_sdk_mcp_server(
                "crash", tools=[open_dump, run_command, list_triages, read_triage]
            )
            try:
                result = Result.model_validate(
                    await run_agent(
                        "Investigate the assigned dump.",
                        {
                            "system_prompt": Path("triage.md").read_text(
                                encoding="utf-8"
                            ),
                            "mcp_servers": {"crash": server},
                            "allowed_tools": [
                                "mcp__crash__" + name
                                for name in (
                                    "open_dump",
                                    "run_command",
                                    "list_triages",
                                    "read_triage",
                                )
                            ],
                            "env": {"MCP_TOOL_TIMEOUT": "200000"},
                            "output_format": {
                                "type": "json_schema",
                                "schema": Result.model_json_schema(),
                            },
                        },
                        on_result,
                        stop=stop,
                    )
                )
                # Require tool evidence before accepting the agent's success or match.
                if result.status == "success" and not (
                    session_id and inspected and listed
                ):
                    raise RuntimeError("Missing debugger evidence or history check.")
                if result.match and (
                    result.status == "failure" or result.match.id not in read_ids
                ):
                    raise RuntimeError("The agent selected an unread or invalid match.")
                return result
            finally:
                if opening:
                    await asyncio.gather(opening, return_exceptions=True)
                if session_id:
                    await call(
                        "close_cdb_session", {"session_id": session_id}, timeout=20
                    )
                    log("debugger.closed", f"session={session_id}")

TypeScript

async function investigate(
  name: string,
  onResult: (result: SDKResultMessage) => void,
) {
  const client = new Client({ name: "crash-triage", version: "1.0.0" });
  const transport = new StreamableHTTPClientTransport(
    new URL(process.env.MCP_WINDBG_URL ?? "http://127.0.0.1:8000/mcp/"),
  );
  let sessionId: string | undefined;
  let opening: Promise<string> | undefined;
  let pending = Promise.resolve();
  let listed = false;
  let inspected = false;
  const readIds = new Set<string>();
  const text = (value: string) => ({
    content: [{ type: "text" as const, text: value }],
  });
  // Serialize debugger calls, and keep the queue usable after a failed call.
  const call = (
    name: string,
    args: Record<string, unknown>,
    timeout = 190_000,
  ) => {
    const operation = pending.then(async () => {
      const reply = await client.callTool(
        { name, arguments: args },
        undefined,
        { timeout },
      );
      const output = (reply.content as { type: string; text?: string }[])
        .filter((x) => x.type === "text")
        .map((x) => x.text)
        .join("\n");
      if (reply.isError) {
        throw new Error(output);
      }
      return output;
    });
    pending = operation.then(
      () => {},
      () => {},
    );
    return operation;
  };
  try {
    log("debugger.connect", "Connecting to MCP WinDbg.");
    await client.connect(transport, { timeout: 10_000 });
    log("debugger.ready", "MCP WinDbg connected.");
    const history = await Promise.all(
      (await readdir("reports/triages", { withFileTypes: true }))
        .filter((f) => f.isFile() && /^[a-f0-9]{16}\.md$/.test(f.name))
        .map(async (f) => ({
          id: f.name.slice(0, -3),
          markdown: await readFile("reports/triages/" + f.name, "utf8"),
        })),
    );
    const server = createSdkMcpServer({
      name: "crash",
      version: "1.0.0",
      alwaysLoad: true,
      timeout: 200_000,
      tools: [
        tool("open_dump", "Open the assigned dump.", {}, async () => {
          // Reuse the session if the agent asks to open the same dump again.
          opening ??= call("open_cdb_dump", {
            dump_path: path.win32.join(process.env.DUMP_HOST_DIR!, name),
            include_stack_trace: true,
            include_modules: true,
            timeout_seconds: 180,
          }).then((output) => {
            sessionId = /^session_id:\s*([a-z]+-[a-z0-9]+)/im.exec(output)?.[1];
            if (!sessionId) {
              throw new Error("No debugger session ID.");
            }
            return output;
          });
          return text(await opening);
        }),
        tool(
          "run_command",
          "Inspect the open dump.",
          {
            command: z.enum(["!analyze -v", ".ecxr", "kv", "kb", "lm", "r"]),
          },
          async ({ command }) => {
            if (!sessionId) {
              throw new Error("Open the dump first.");
            }
            const output = await call("run_cdb_command", {
              session_id: sessionId,
              command,
            });
            inspected = true;
            return text(output);
          },
        ),
        tool("list_triages", "Find earlier triages.", {}, async () => {
          listed = true;
          return text(
            JSON.stringify(
              history.map((h) => ({
                id: h.id,
                summary: h.markdown.split("\n\n")[1],
              })),
            ),
          );
        }),
        tool(
          "read_triage",
          "Read a possible match.",
          { id: z.string() },
          async ({ id }) => {
            const entry = history.find((h) => h.id === id);
            if (!entry) {
              throw new Error("Unknown triage.");
            }
            readIds.add(id);
            return text(entry.markdown);
          },
        ),
      ],
    });
    const result = Result.parse(
      await runAgent(
        "Investigate the assigned dump.",
        {
          systemPrompt: await readFile("triage.md", "utf8"),
          mcpServers: { crash: server },
          allowedTools: [
            "open_dump",
            "run_command",
            "list_triages",
            "read_triage",
          ].map((name) => "mcp__crash__" + name),
          abortController: stop,
          outputFormat: {
            type: "json_schema",
            schema: z.toJSONSchema(Result, { target: "draft-7" }),
          },
        },
        onResult,
      ),
    );
    // Require tool evidence before accepting the agent's success or match.
    if (result.status === "success" && (!sessionId || !inspected || !listed)) {
      throw new Error("Missing debugger evidence or history check.");
    }
    if (
      result.match &&
      (result.status === "failure" || !readIds.has(result.match.id))
    ) {
      throw new Error("The agent selected an unread or invalid match.");
    }
    return result;
  } finally {
    // Finish any pending command before closing its debugger session.
    await pending;
    try {
      if (sessionId) {
        await call("close_cdb_session", { session_id: sessionId }, 20_000);
        log("debugger.closed", `session=${sessionId}`);
      }
    } finally {
      try {
        if (transport.sessionId) {
          await transport.terminateSession();
        }
      } finally {
        await client.close();
      }
    }
  }
}

Python

def save(name, result: StoredResult):
    occurrence_id = hashlib.sha256(name.encode("utf-8")).hexdigest()[:16]
    group = (
        (result.match.id if result.match else occurrence_id)
        if result.status == "success"
        else None
    )
    if group:
        file = Path("reports/triages") / f"{group}.md"
        try:
            previous = file.read_text(encoding="utf-8")
        except FileNotFoundError:
            if result.match:
                raise
            previous = f"# Triage {group}\n\n{result.summary}\n\n{result.report}\n\n## Occurrences\n"
        # Avoid duplicate occurrences if a run stops after updating this triage.
        marker = f"<!-- occurrence:{occurrence_id} -->"
        if marker not in previous:
            reason = result.match.reason if result.match else "First occurrence."
            file.write_text(
                previous
                + f"\n{marker}\n- [{name}](../{quote(name, safe='')}.md): {reason}\n",
                encoding="utf-8",
            )
    markdown = f"# {name}\n\nStatus: **{result.status}**\n\n{result.report}\n"
    if group:
        markdown += f"\n[Triage {group}](triages/{group}.md)\n"
    (Path("reports") / f"{name}.md").write_text(markdown, encoding="utf-8")
    # Write JSON last: its presence marks this input as completed.
    (Path("reports") / f"{name}.json").write_text(
        result.model_dump_json(indent=2), encoding="utf-8"
    )

TypeScript

async function save(name: string, result: z.infer<typeof StoredResult>) {
  const id = createHash("sha256").update(name).digest("hex").slice(0, 16);
  const group = result.status === "success" ? (result.match?.id ?? id) : null;
  if (group) {
    const file = "reports/triages/" + group + ".md";
    const previous = await readFile(file, "utf8").catch((error) => {
      if (error.code !== "ENOENT" || result.match) {
        throw error;
      }
      return `# Triage ${group}\n\n${result.summary}\n\n${result.report}\n\n## Occurrences\n`;
    });
    // Avoid duplicate occurrences if a run stops after updating this triage.
    const marker = `<!-- occurrence:${id} -->`;
    if (!previous.includes(marker)) {
      await writeFile(
        file,
        previous +
          `\n${marker}\n- [${name}](../${encodeURIComponent(name)}.md): ${result.match?.reason ?? "First occurrence."}\n`,
      );
    }
  }
  await writeFile(
    "reports/" + name + ".md",
    `# ${name}\n\nStatus: **${result.status}**\n\n${result.report}\n` +
      (group ? `\n[Triage ${group}](triages/${group}.md)\n` : ""),
  );
  // Write JSON last: its presence marks this input as completed.
  await writeFile("reports/" + name + ".json", JSON.stringify(result, null, 2));
}

Python

async def main():
    if not ntpath.isabs(os.environ.get("DUMP_HOST_DIR", "")):
        raise ValueError("Set DUMP_HOST_DIR to the full Windows path of dumps.")
    once = "--once" in sys.argv
    log(
        "worker.start",
        f"pid={os.getpid()} mode={'once' if once else 'watch'} dumps={Path('dumps').resolve()} reports={Path('reports').resolve()}",
    )
    waiting = False
    exit_code = 0
    try:
        while not stop.is_set():
            for file in sorted(Path("dumps").iterdir()):
                if stop.is_set():
                    break
                if not file.is_file() or file.suffix.lower() != ".dmp":
                    continue

                # Reuse completed results so polling doesn't start another paid session.
                saved = Path("reports") / f"{file.name}.json"
                if saved.exists():
                    result = StoredResult.model_validate_json(
                        saved.read_text(encoding="utf-8")
                    )
                else:
                    waiting = False
                    started = monotonic()
                    log("dump.start", file.name)
                    estimated_cost = None

                    def on_result(message):
                        nonlocal estimated_cost
                        estimated_cost = message.total_cost_usd

                    try:
                        triage = await investigate(file.name, on_result)
                    except Exception as error:
                        detail = f"{type(error).__name__}: {error}"
                        triage = Result(
                            status="failure",
                            summary=detail,
                            report=detail,
                            match=None,
                        )
                    result = StoredResult(
                        **triage.model_dump(), estimatedCostUsd=estimated_cost
                    )
                    save(file.name, result)
                    log(
                        "dump.stop",
                        f"{file.name} status={result.status} elapsedMs={int((monotonic() - started) * 1000)}",
                    )
                    log("dump.summary", result.summary)
                    log(
                        "dump.saved",
                        f"reports/{file.name}.json and reports/{file.name}.md",
                    )
                if result.status == "failure":
                    exit_code = 1
            if once:
                break
            if not waiting and not stop.is_set():
                log(
                    "worker.waiting",
                    "All current dumps processed; checking for new files every 2s.",
                )
                waiting = True
            try:
                await asyncio.wait_for(stop.wait(), timeout=2)
            except TimeoutError:
                pass
    finally:
        log("worker.stop", f"exitCode={exit_code}")
    return exit_code


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))

TypeScript

if (
  !process.env.DUMP_HOST_DIR ||
  !path.win32.isAbsolute(process.env.DUMP_HOST_DIR)
) {
  throw new Error("Set DUMP_HOST_DIR to the full Windows path of dumps.");
}
log(
  "worker.start",
  `pid=${process.pid} mode=${process.argv.includes("--once") ? "once" : "watch"} dumps=${path.resolve("dumps")} reports=${path.resolve("reports")}`,
);
let waiting = false;
do {
  for (const file of await readdir("dumps", { withFileTypes: true })) {
    if (stop.signal.aborted) {
      break;
    }
    if (!file.isFile() || !/\.dmp$/i.test(file.name)) {
      continue;
    }

    // Reuse completed results so polling doesn't start another paid session.
    const saved = await readFile(
      "reports/" + file.name + ".json",
      "utf8",
    ).catch((error) => {
      if (error.code !== "ENOENT") {
        throw error;
      }
      return null;
    });
    const started = Date.now();
    if (!saved) {
      waiting = false;
      log("dump.start", file.name);
    }
    let estimatedCostUsd: number | null = null;
    const result = saved
      ? StoredResult.parse(JSON.parse(saved))
      : {
          ...(await investigate(file.name, (message) => {
            estimatedCostUsd = message.total_cost_usd;
          }).catch((error) => ({
            status: "failure" as const,
            summary: String(error),
            report: String(error),
            match: null,
          }))),
          estimatedCostUsd,
        };
    if (!saved) {
      await save(file.name, result);
    }
    if (result.status === "failure") {
      process.exitCode = 1;
    }
    if (!saved) {
      log(
        "dump.stop",
        `${file.name} status=${result.status} elapsedMs=${Date.now() - started}`,
      );
      log("dump.summary", result.summary);
      log(
        "dump.saved",
        `reports/${file.name}.json and reports/${file.name}.md`,
      );
    }
  }
  if (process.argv.includes("--once")) {
    break;
  }
  if (!waiting && !stop.signal.aborted) {
    log(
      "worker.waiting",
      "All current dumps processed; checking for new files every 2s.",
    );
    waiting = true;
  }
  await sleep(2_000, undefined, { signal: stop.signal }).catch(() => {});
} while (!stop.signal.aborted);

Python

DUMP_HOST_DIR='C:\Code\agent-sdk-starter-python\dumps'

TypeScript

DUMP_HOST_DIR='C:\Code\agent-sdk-starter-typescript\dumps'

Python

services:
  agent:
    build: .
    init: true
    env_file: .env
    environment:
      MCP_WINDBG_URL: http://host.docker.internal:8000/mcp/
    command: ["uv", "run", "--no-sync", "triage.py"]
    volumes:
      - ./dumps:/app/dumps:ro
      - ./reports:/app/reports
      - ./triage.md:/app/triage.md:ro
    stop_grace_period: 4m

TypeScript

services:
  agent:
    build: .
    init: true
    env_file: .env
    environment:
      MCP_WINDBG_URL: http://host.docker.internal:8000/mcp/
    command: ["node", "triage.ts"]
    volumes:
      - ./dumps:/app/dumps:ro
      - ./reports:/app/reports
      - ./triage.md:/app/triage.md:ro
    stop_grace_period: 4m