#!/usr/bin/env python3 """Same task, three tool surfaces: MCP server, bare CLI, CLI plus a SKILL.md. One file holds everything so the comparison is reproducible: * ``seed`` create ``issues.sqlite3`` with a fixed set of issues * ``cli ...`` a small issue-tracker command line (what the "CLI" and "skill" modes drive) * ``mcp-server`` the same operations exposed as MCP tools over stdio * ``run`` drive Claude through the task N times per mode and print metrics * ``verify`` check the database against the expected end state Requirements: Python 3.10+, ``pip install "anthropic>=1.6" "mcp>=2,<3"``, ``ANTHROPIC_API_KEY``. python tool_surface_comparison.py run --mode mcp --trials 5 python tool_surface_comparison.py run --mode cli --trials 5 python tool_surface_comparison.py run --mode skill --trials 5 """ from __future__ import annotations import argparse import asyncio import json import os import sqlite3 import subprocess import sys import time from pathlib import Path DB = Path("issues.sqlite3") TODAY = "2026-09-17" MODEL = "claude-opus-5" TASK = ( f"Today is {TODAY}. In our issue tracker, close every OPEN issue assigned to 'kim' " "that has not been updated for more than 30 days. Use exactly the comment " "'Closed for inactivity'. Do not change any other issue. When done, list the ids you closed." ) # id, title, assignee, state, updated_at, comment SEED = [ (101, "Login page 500 on empty password", "kim", "open", "2026-06-02", ""), (102, "Export CSV drops header row", "kim", "open", "2026-07-30", ""), (103, "Dark mode contrast too low", "kim", "open", "2026-09-10", ""), # fresh: keep (104, "Search ignores accents", "kim", "closed", "2026-05-01", "duplicate"), # already closed (105, "Retry storm on 429", "kimberly", "open", "2026-05-20", ""), # different person (106, "Slow dashboard query", "lee", "open", "2026-04-11", ""), (107, "Typo in onboarding email", "kim", "open", "2026-08-01", ""), (108, "Webhook signature mismatch", "kim", "open", "2026-08-25", ""), # 23 days: keep ] EXPECTED_CLOSED = {101, 102, 107} SKILL_MD = """--- name: issue-tracker-cli description: Query and close issues with the `issues` command. Use when asked to list, inspect, or close tracker issues. --- # issues CLI Every command prints JSON when given `--json`. Always use `--json`. - `issues list --json [--assignee NAME] [--state open|closed]` lists issues with `id`, `title`, `assignee`, `state`, `updated_at` (YYYY-MM-DD). - `issues show ID --json` prints one issue. - `issues close ID --comment "TEXT"` closes one issue. Refuses to close a closed issue. Rules: 1. Filter server-side with `--assignee` and `--state`; the assignee match is exact. 2. Compute "days since update" yourself from `updated_at` and the date in the task. 3. Close issues one command at a time and read each response. """ # --- storage ------------------------------------------------------------------ def connect() -> sqlite3.Connection: conn = sqlite3.connect(DB) conn.row_factory = sqlite3.Row return conn def seed() -> None: if DB.exists(): DB.unlink() with connect() as conn: conn.execute( "CREATE TABLE issues (id INTEGER PRIMARY KEY, title TEXT, assignee TEXT, " "state TEXT, updated_at TEXT, comment TEXT)" ) conn.executemany("INSERT INTO issues VALUES (?, ?, ?, ?, ?, ?)", SEED) def list_issues(assignee: str | None = None, state: str | None = None) -> list[dict]: sql, params = "SELECT * FROM issues WHERE 1=1", [] if assignee: sql += " AND assignee = ?" params.append(assignee) if state: sql += " AND state = ?" params.append(state) with connect() as conn: return [dict(r) for r in conn.execute(sql + " ORDER BY id", params)] def get_issue(issue_id: int) -> dict: with connect() as conn: row = conn.execute("SELECT * FROM issues WHERE id = ?", (issue_id,)).fetchone() if row is None: raise ValueError(f"issue {issue_id} not found") return dict(row) def close_issue(issue_id: int, comment: str) -> dict: issue = get_issue(issue_id) if issue["state"] == "closed": raise ValueError(f"issue {issue_id} is already closed") with connect() as conn: conn.execute( "UPDATE issues SET state = 'closed', comment = ?, updated_at = ? WHERE id = ?", (comment, TODAY, issue_id), ) return get_issue(issue_id) def expected_rows() -> dict[int, tuple]: rows = {} for issue_id, title, assignee, state, updated_at, comment in SEED: if issue_id in EXPECTED_CLOSED: rows[issue_id] = (issue_id, title, assignee, "closed", TODAY, "Closed for inactivity") else: rows[issue_id] = (issue_id, title, assignee, state, updated_at, comment) return rows def verify() -> list[str]: """Compare the whole table with the expected end state; empty list means exact match.""" actual = { r["id"]: (r["id"], r["title"], r["assignee"], r["state"], r["updated_at"], r["comment"]) for r in list_issues() } expected = expected_rows() problems = [] for issue_id in sorted(set(actual) | set(expected)): if issue_id not in expected: problems.append(f"{issue_id}: unexpected row {actual[issue_id]}") elif issue_id not in actual: problems.append(f"{issue_id}: row missing") elif actual[issue_id] != expected[issue_id]: problems.append(f"{issue_id}: got {actual[issue_id]}, expected {expected[issue_id]}") return problems # --- CLI surface ---------------------------------------------------------------- def cli(argv: list[str]) -> int: parser = argparse.ArgumentParser(prog="issues", description="Tiny issue tracker.") sub = parser.add_subparsers(dest="cmd", required=True) p_list = sub.add_parser("list", help="list issues") p_list.add_argument("--assignee") p_list.add_argument("--state", choices=["open", "closed"]) p_list.add_argument("--json", action="store_true") p_show = sub.add_parser("show", help="show one issue") p_show.add_argument("id", type=int) p_show.add_argument("--json", action="store_true") p_close = sub.add_parser("close", help="close an issue with a comment") p_close.add_argument("id", type=int) p_close.add_argument("--comment", required=True) args = parser.parse_args(argv) try: if args.cmd == "list": rows = list_issues(args.assignee, args.state) if args.json: print(json.dumps(rows)) else: for r in rows: print(f"#{r['id']:<4} {r['state']:<6} {r['assignee']:<9} {r['updated_at']} {r['title']}") elif args.cmd == "show": row = get_issue(args.id) print(json.dumps(row) if args.json else "\n".join(f"{k}: {v}" for k, v in row.items())) else: row = close_issue(args.id, args.comment) print(f"closed #{row['id']}") except ValueError as exc: print(f"error: {exc}", file=sys.stderr) return 1 return 0 # --- MCP surface ------------------------------------------------------------------ def mcp_server() -> None: from mcp.server import MCPServer # noqa: PLC0415 from mcp.server.mcpserver.exceptions import ToolError # noqa: PLC0415 server = MCPServer("issue-tracker") def guarded(fn): # A plain exception reaches the model as "Error executing tool X" with the message dropped. # ToolError keeps the message, which is what the model needs to recover. def wrapper(**kwargs): try: return fn(**kwargs) except ValueError as exc: raise ToolError(str(exc)) from exc wrapper.__name__, wrapper.__doc__, wrapper.__annotations__ = fn.__name__, fn.__doc__, fn.__annotations__ wrapper.__signature__ = __import__("inspect").signature(fn) return wrapper @server.tool() @guarded def list_issues_tool(assignee: str | None = None, state: str | None = None) -> list[dict]: """List issues. Filters are exact matches. `updated_at` is YYYY-MM-DD.""" return list_issues(assignee, state) @server.tool() @guarded def get_issue_tool(issue_id: int) -> dict: """Get one issue by id.""" return get_issue(issue_id) @server.tool() @guarded def close_issue_tool(issue_id: int, comment: str) -> dict: """Close an open issue with a comment. Fails if it is already closed.""" return close_issue(issue_id, comment) server.run() # --- the agent loop shared by all modes ------------------------------------------- def bash_tool(bin_dir: Path): def run(command: str) -> str: proc = subprocess.run( command, shell=True, capture_output=True, text=True, timeout=30, env={**os.environ, "PATH": f"{bin_dir}:{os.environ['PATH']}"}, ) out = proc.stdout + (("\n[stderr]\n" + proc.stderr) if proc.stderr else "") return f"exit={proc.returncode}\n{out}"[:8000] return run BASH_TOOL_DEF = { "name": "bash", "description": "Run a shell command and return its exit code and output.", "input_schema": { "type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"], "additionalProperties": False, }, "strict": True, } async def trial(mode: str, bin_dir: Path) -> dict: import anthropic # noqa: PLC0415 client = anthropic.Anthropic() system = "You are an operations assistant. Finish the task with the tools available, then report." tools: list[dict] dispatch: dict mcp_client = None if mode == "mcp": from mcp import Client # noqa: PLC0415 from mcp.client.stdio import StdioServerParameters # noqa: PLC0415 mcp_client = Client(StdioServerParameters(command=sys.executable, args=[__file__, "mcp-server"])) await mcp_client.__aenter__() listed = await mcp_client.list_tools() tools = [ {"name": t.name, "description": t.description or "", "input_schema": t.input_schema} for t in listed.tools ] async def call_mcp(name: str, args: dict) -> str: result = await mcp_client.call_tool(name, args) text = "\n".join(c.text for c in result.content if getattr(c, "type", "") == "text") if result.is_error: raise ValueError(text) return text dispatch = {t["name"]: call_mcp for t in tools} else: tools = [BASH_TOOL_DEF] run_bash = bash_tool(bin_dir) async def call_bash(name: str, args: dict) -> str: return run_bash(args["command"]) dispatch = {"bash": call_bash} system += " An `issues` command line tool is installed." if mode == "skill": system += "\n\nA skill is active. Follow it:\n\n" + SKILL_MD schema_tokens = client.messages.count_tokens(model=MODEL, tools=tools, system=system, messages=[{"role": "user", "content": "x"}]).input_tokens messages = [{"role": "user", "content": TASK}] stats = {"mode": mode, "calls": 0, "tool_calls": 0, "tool_errors": 0, "input_tokens": 0, "output_tokens": 0, "last_input_tokens": 0, "static_tokens": schema_tokens, "commands": []} started = time.time() try: while True: response = client.messages.create(model=MODEL, max_tokens=8000, system=system, tools=tools, messages=messages) stats["calls"] += 1 stats["input_tokens"] += response.usage.input_tokens stats["output_tokens"] += response.usage.output_tokens stats["last_input_tokens"] = response.usage.input_tokens messages.append({"role": "assistant", "content": [b.model_dump(exclude_none=True) for b in response.content]}) if response.stop_reason != "tool_use": stats["final"] = "".join(b.text for b in response.content if b.type == "text")[:400] break results = [] for block in response.content: if block.type != "tool_use": continue stats["tool_calls"] += 1 label = block.input.get("command") if block.name == "bash" else f"{block.name}({json.dumps(block.input)})" stats["commands"].append(label) try: content = await dispatch[block.name](block.name, block.input) results.append({"type": "tool_result", "tool_use_id": block.id, "content": content}) except Exception as exc: stats["tool_errors"] += 1 results.append({"type": "tool_result", "tool_use_id": block.id, "content": f"Error: {exc}", "is_error": True}) messages.append({"role": "user", "content": results}) if stats["calls"] > 25: stats["final"] = "aborted: too many calls" break finally: if mcp_client is not None: await mcp_client.__aexit__(None, None, None) stats["seconds"] = round(time.time() - started, 1) stats["problems"] = verify() stats["success"] = not stats["problems"] return stats def run_trials(mode: str, trials: int) -> None: bin_dir = Path(".issues-bin").resolve() bin_dir.mkdir(exist_ok=True) shim = bin_dir / "issues" shim.write_text(f'#!/bin/sh\nexec "{sys.executable}" "{Path(__file__).resolve()}" cli "$@"\n') shim.chmod(0o755) results = [] for i in range(trials): seed() stats = asyncio.run(trial(mode, bin_dir)) results.append(stats) print(f"[{mode} trial {i + 1}] success={stats['success']} calls={stats['calls']} " f"tool_calls={stats['tool_calls']} errors={stats['tool_errors']} " f"in={stats['input_tokens']} out={stats['output_tokens']} " f"last_in={stats['last_input_tokens']} static={stats['static_tokens']} {stats['seconds']}s") if stats["problems"]: print(" problems:", stats["problems"]) for command in stats["commands"]: print(" -", command[:140]) Path(f"results_{mode}.json").write_text(json.dumps(results, indent=2, ensure_ascii=False)) ok = sum(r["success"] for r in results) mean = lambda key: sum(r[key] for r in results) / len(results) # noqa: E731 print(f"\n{mode}: {ok}/{trials} correct | mean calls {mean('calls'):.1f} | tool calls {mean('tool_calls'):.1f} " f"| input tokens {mean('input_tokens'):,.0f} | output tokens {mean('output_tokens'):,.0f} " f"| final-call context {mean('last_input_tokens'):,.0f} | static {results[0]['static_tokens']:,} | {mean('seconds'):.1f}s") def main() -> None: if len(sys.argv) > 1 and sys.argv[1] == "cli": sys.exit(cli(sys.argv[2:])) parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) sub = parser.add_subparsers(dest="cmd", required=True) sub.add_parser("seed") sub.add_parser("mcp-server") sub.add_parser("verify") p_run = sub.add_parser("run") p_run.add_argument("--mode", choices=["mcp", "cli", "skill"], required=True) p_run.add_argument("--trials", type=int, default=3) args = parser.parse_args() if args.cmd == "seed": seed() print(f"seeded {DB} with {len(SEED)} issues") elif args.cmd == "mcp-server": mcp_server() elif args.cmd == "verify": problems = verify() print("OK" if not problems else "\n".join(problems)) sys.exit(1 if problems else 0) else: run_trials(args.mode, args.trials) if __name__ == "__main__": main()