The Tool Interface — Why Agents Need Structured I/O

"""Phase 13 Lesson 01 - the tool interface, four-step loop, no LLM.

Implements the describe -> decide -> execute -> observe cycle used by every
2026 tool-calling stack (OpenAI, Anthropic, Gemini, MCP, A2A). The "decide"
step is faked with a keyword router so the loop runs offline; replace it with
any real provider in Lesson 02.

The harness:
  - registers three tools (add, get_time, get_weather)
  - validates tool-call arguments against a minimal JSON Schema subset
  - prints each step so you can read the choreography
  - bounds iteration at MAX_TURNS to prevent runaway loops

Run: python code/main.py
"""

from __future__ import annotations

import datetime as dt
import json
import re
import time
import uuid
from dataclasses import dataclass
from typing import Any, Callable


MAX_TURNS = 5


@dataclass
class Tool:
    name: str
    description: str
    input_schema: dict
    executor: Callable[[dict], Any]
    consequential: bool = False


def tool_add(args: dict) -> dict:
    return {"sum": args["a"] + args["b"]}


def tool_get_time(args: dict) -> dict:
    tz = args.get("timezone", "UTC")
    now = dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds")
    return {"now": now, "timezone": tz}


def tool_get_weather(args: dict) -> dict:
    fake = {"Bengaluru": 28, "Tokyo": 12, "Zurich": 4, "Lagos": 31}
    city = args["city"]
    units = args.get("units", "celsius")
    temp = fake.get(city, 20)
    return {"city": city, "temp": temp, "units": units}


REGISTRY: list[Tool] = [
    Tool(
        name="add",
        description=(
            "Use when the user asks for the sum of two numbers. "
            "Do not use for subtraction, product, or symbolic algebra."
        ),
        input_schema={
            "type": "object",
            "properties": {
                "a": {"type": "number"},
                "b": {"type": "number"},
            },
            "required": ["a", "b"],
        },
        executor=tool_add,
    ),
    Tool(
        name="get_time",
        description=(
            "Use when the user asks what time it is. "
            "Do not use for historical dates or future scheduling."
        ),
        input_schema={
            "type": "object",
            "properties": {
                "timezone": {"type": "string"},
            },
            "required": [],
        },
        executor=tool_get_time,
    ),
    Tool(
        name="get_weather",
        description=(
            "Use when the user asks about current conditions in a named city. "
            "Do not use for forecasts or historical weather data."
        ),
        input_schema={
            "type": "object",
            "properties": {
                "city": {"type": "string"},
                "units": {"type": "string", "enum": ["celsius", "fahrenheit"]},
            },
            "required": ["city"],
        },
        executor=tool_get_weather,
    ),
]


def validate(schema: dict, value: Any) -> list[str]:
    errors: list[str] = []
    t = schema.get("type")
    if t == "object":
        if not isinstance(value, dict):
            return [f"expected object, got {type(value).__name__}"]
        for field in schema.get("required", []):
            if field not in value:
                errors.append(f"missing required field '{field}'")
        for key, sub in schema.get("properties", {}).items():
            if key in value:
                errors.extend(validate(sub, value[key]))
        return errors
    if t == "number" and not isinstance(value, (int, float)):
        errors.append(f"expected number, got {type(value).__name__}")
    if t == "string" and not isinstance(value, str):
        errors.append(f"expected string, got {type(value).__name__}")
    if "enum" in schema and value not in schema["enum"]:
        errors.append(f"value {value!r} not in enum {schema['enum']}")
    return errors


def fake_decide(user_msg: str, history: list[dict]) -> dict:
    """Stand-in for the model. Routes by keyword so the loop runs offline.

    Production substitute: swap this for provider.chat.completions.create with
    tools=[t.input_schema for t in REGISTRY]. Same return shape.
    """
    last = history[-1] if history else {}
    if last.get("role") == "tool":
        return {"content": f"Final answer built from tool output: {last.get('content')}"}
    msg = user_msg.lower()
    if re.search(r"\b(add|sum|plus)\b", msg):
        nums = [float(n) for n in re.findall(r"-?\d+\.?\d*", msg)]
        if len(nums) >= 2:
            return {
                "tool_calls": [
                    {
                        "id": f"call_{uuid.uuid4().hex[:8]}",
                        "name": "add",
                        "arguments": {"a": nums[0], "b": nums[1]},
                    }
                ]
            }
    if "time" in msg:
        return {
            "tool_calls": [
                {
                    "id": f"call_{uuid.uuid4().hex[:8]}",
                    "name": "get_time",
                    "arguments": {"timezone": "UTC"},
                }
            ]
        }
    match = re.search(r"weather in (\w+)", msg)
    if match:
        city = match.group(1).title()
        return {
            "tool_calls": [
                {
                    "id": f"call_{uuid.uuid4().hex[:8]}",
                    "name": "get_weather",
                    "arguments": {"city": city, "units": "celsius"},
                }
            ]
        }
    return {"content": "I cannot route that query to any registered tool."}


def run_loop(user_msg: str) -> None:
    print("=" * 72)
    print(f"USER : {user_msg}")
    print("-" * 72)
    tools_by_name = {t.name: t for t in REGISTRY}
    history: list[dict] = [{"role": "user", "content": user_msg}]
    for turn in range(1, MAX_TURNS + 1):
        decision = fake_decide(user_msg, history)
        if "content" in decision:
            print(f"TURN {turn} DECIDE : final answer")
            print(f"MODEL : {decision['content']}")
            return
        for call in decision["tool_calls"]:
            tool = tools_by_name.get(call["name"])
            print(f"TURN {turn} DECIDE : call {call['name']} id={call['id']}")
            print(f"           args = {json.dumps(call['arguments'])}")
            if tool is None:
                print(f"           ERROR : unknown tool {call['name']}")
                return
            errs = validate(tool.input_schema, call["arguments"])
            if errs:
                print(f"           VALIDATION ERRORS : {errs}")
                return
            if tool.consequential:
                print("           GATE : tool is consequential, would confirm")
            start = time.perf_counter()
            result = tool.executor(call["arguments"])
            ms = (time.perf_counter() - start) * 1000
            print(f"TURN {turn} EXECUTE: {tool.name} -> {json.dumps(result)}"
                  f" [{ms:.2f} ms]")
            history.append({
                "role": "tool", "id": call["id"],
                "name": tool.name, "content": json.dumps(result),
            })
        print(f"TURN {turn} OBSERVE: history length = {len(history)}")
    print("LOOP TERMINATED : hit MAX_TURNS circuit breaker")


def describe_registry() -> None:
    print("TOOL REGISTRY")
    print("-" * 72)
    for t in REGISTRY:
        kind = "consequential" if t.consequential else "pure"
        print(f"  {t.name:14s} [{kind}] - {t.description}")
    print()


def main() -> None:
    print("=" * 72)
    print("PHASE 13 LESSON 01 - THE TOOL INTERFACE")
    print("=" * 72)
    describe_registry()
    for query in (
        "please add 7 and 35",
        "what time is it?",
        "tell me the weather in Bengaluru",
        "write me a haiku about tea",
    ):
        run_loop(query)
        print()


if __name__ == "__main__":
    main()

// Phase 13 Lesson 01 — the tool interface, in TypeScript.
//
// Mirrors code/main.py: describe -> decide -> execute -> observe.
// The "decide" step is faked with a keyword router so the loop runs offline;
// replace with any real provider client and the shape stays the same.
//
// Spec references:
//   OpenAI tool calling     https://platform.openai.com/docs/guides/function-calling
//   Anthropic tool use      https://docs.anthropic.com/en/docs/build-with-claude/tool-use
//   MCP tool primitive      https://modelcontextprotocol.io/specification/2025-11-25
//
// Run: npx tsx code/main.ts

import { randomUUID } from "node:crypto";

const MAX_TURNS = 5;

type JsonSchema = {
  type?: "object" | "string" | "number" | "integer" | "boolean" | "array";
  properties?: Record<string, JsonSchema>;
  required?: string[];
  enum?: unknown[];
};

type ToolArgs = Record<string, unknown>;
type ToolResult = Record<string, unknown>;

type Tool = {
  name: string;
  description: string;
  inputSchema: JsonSchema;
  executor: (args: ToolArgs) => ToolResult;
  consequential?: boolean;
};

type HistoryEntry =
  | { role: "user"; content: string }
  | { role: "tool"; id: string; name: string; content: string };

type ToolCall = {
  id: string;
  name: string;
  arguments: ToolArgs;
};

type Decision = { content: string } | { toolCalls: ToolCall[] };

function toolAdd(args: ToolArgs): ToolResult {
  const a = args.a as number;
  const b = args.b as number;
  return { sum: a + b };
}

function toolGetTime(args: ToolArgs): ToolResult {
  const timezone = (args.timezone as string | undefined) ?? "UTC";
  const now = new Date().toISOString().replace(/\.\d{3}Z$/, "Z");
  return { now, timezone };
}

function toolGetWeather(args: ToolArgs): ToolResult {
  const fake: Record<string, number> = {
    Bengaluru: 28,
    Tokyo: 12,
    Zurich: 4,
    Lagos: 31,
  };
  const city = args.city as string;
  const units = (args.units as string | undefined) ?? "celsius";
  const temp = fake[city] ?? 20;
  return { city, temp, units };
}

const REGISTRY: Tool[] = [
  {
    name: "add",
    description:
      "Use when the user asks for the sum of two numbers. " +
      "Do not use for subtraction, product, or symbolic algebra.",
    inputSchema: {
      type: "object",
      properties: {
        a: { type: "number" },
        b: { type: "number" },
      },
      required: ["a", "b"],
    },
    executor: toolAdd,
  },
  {
    name: "get_time",
    description:
      "Use when the user asks what time it is. " +
      "Do not use for historical dates or future scheduling.",
    inputSchema: {
      type: "object",
      properties: {
        timezone: { type: "string" },
      },
      required: [],
    },
    executor: toolGetTime,
  },
  {
    name: "get_weather",
    description:
      "Use when the user asks about current conditions in a named city. " +
      "Do not use for forecasts or historical weather data.",
    inputSchema: {
      type: "object",
      properties: {
        city: { type: "string" },
        units: { type: "string", enum: ["celsius", "fahrenheit"] },
      },
      required: ["city"],
    },
    executor: toolGetWeather,
  },
];

function validate(schema: JsonSchema, value: unknown): string[] {
  const errors: string[] = [];
  const t = schema.type;

  if (t === "object") {
    if (typeof value !== "object" || value === null || Array.isArray(value)) {
      return [`expected object, got ${describeType(value)}`];
    }
    const obj = value as Record<string, unknown>;
    for (const field of schema.required ?? []) {
      if (!(field in obj)) errors.push(`missing required field '${field}'`);
    }
    for (const [key, sub] of Object.entries(schema.properties ?? {})) {
      if (key in obj) errors.push(...validate(sub, obj[key]));
    }
    return errors;
  }

  if (t === "number" && typeof value !== "number") {
    errors.push(`expected number, got ${describeType(value)}`);
  }
  if (t === "string" && typeof value !== "string") {
    errors.push(`expected string, got ${describeType(value)}`);
  }
  if (schema.enum && !schema.enum.includes(value as never)) {
    errors.push(`value ${JSON.stringify(value)} not in enum ${JSON.stringify(schema.enum)}`);
  }
  return errors;
}

function describeType(value: unknown): string {
  if (value === null) return "null";
  if (Array.isArray(value)) return "array";
  return typeof value;
}

function newCallId(): string {
  return `call_${randomUUID().replace(/-/g, "").slice(0, 8)}`;
}

// Stand-in for the model. Routes by keyword so the loop runs offline.
// Production substitute: replace with a provider call returning the same shape.
function fakeDecide(userMsg: string, history: HistoryEntry[]): Decision {
  const last = history[history.length - 1];
  if (last && last.role === "tool") {
    return { content: `Final answer built from tool output: ${last.content}` };
  }
  const msg = userMsg.toLowerCase();

  if (/\b(add|sum|plus)\b/.test(msg)) {
    const nums = (msg.match(/-?\d+\.?\d*/g) ?? []).map((n) => Number(n));
    if (nums.length >= 2) {
      return {
        toolCalls: [
          { id: newCallId(), name: "add", arguments: { a: nums[0], b: nums[1] } },
        ],
      };
    }
  }

  if (msg.includes("time")) {
    return {
      toolCalls: [
        { id: newCallId(), name: "get_time", arguments: { timezone: "UTC" } },
      ],
    };
  }

  const weatherMatch = msg.match(/weather in (\w+)/);
  if (weatherMatch) {
    const city = weatherMatch[1][0].toUpperCase() + weatherMatch[1].slice(1);
    return {
      toolCalls: [
        {
          id: newCallId(),
          name: "get_weather",
          arguments: { city, units: "celsius" },
        },
      ],
    };
  }

  return { content: "I cannot route that query to any registered tool." };
}

function runLoop(userMsg: string): void {
  console.log("=".repeat(72));
  console.log(`USER : ${userMsg}`);
  console.log("-".repeat(72));

  const toolsByName = new Map(REGISTRY.map((t) => [t.name, t]));
  const history: HistoryEntry[] = [{ role: "user", content: userMsg }];

  for (let turn = 1; turn <= MAX_TURNS; turn++) {
    const decision = fakeDecide(userMsg, history);

    if ("content" in decision) {
      console.log(`TURN ${turn} DECIDE : final answer`);
      console.log(`MODEL : ${decision.content}`);
      return;
    }

    for (const call of decision.toolCalls) {
      const tool = toolsByName.get(call.name);
      console.log(`TURN ${turn} DECIDE : call ${call.name} id=${call.id}`);
      console.log(`           args = ${JSON.stringify(call.arguments)}`);

      if (!tool) {
        console.log(`           ERROR : unknown tool ${call.name}`);
        return;
      }
      const errs = validate(tool.inputSchema, call.arguments);
      if (errs.length > 0) {
        console.log(`           VALIDATION ERRORS : ${JSON.stringify(errs)}`);
        return;
      }
      if (tool.consequential) {
        console.log("           GATE : tool is consequential, would confirm");
      }

      const start = performance.now();
      const result = tool.executor(call.arguments);
      const ms = performance.now() - start;
      console.log(
        `TURN ${turn} EXECUTE: ${tool.name} -> ${JSON.stringify(result)} [${ms.toFixed(2)} ms]`,
      );
      history.push({
        role: "tool",
        id: call.id,
        name: tool.name,
        content: JSON.stringify(result),
      });
    }
    console.log(`TURN ${turn} OBSERVE: history length = ${history.length}`);
  }
  console.log("LOOP TERMINATED : hit MAX_TURNS circuit breaker");
}

function describeRegistry(): void {
  console.log("TOOL REGISTRY");
  console.log("-".repeat(72));
  for (const t of REGISTRY) {
    const kind = t.consequential ? "consequential" : "pure";
    console.log(`  ${t.name.padEnd(14)} [${kind}] - ${t.description}`);
  }
  console.log();
}

function main(): void {
  console.log("=".repeat(72));
  console.log("PHASE 13 LESSON 01 - THE TOOL INTERFACE (TypeScript port)");
  console.log("=".repeat(72));
  describeRegistry();
  const queries = [
    "please add 7 and 35",
    "what time is it?",
    "tell me the weather in Bengaluru",
    "write me a haiku about tea",
  ];
  for (const q of queries) {
    runLoop(q);
    console.log();
  }
}

main();