L2 · 04 · Tool Calling from Your Code

In Level 1 you used agents. Now you build the core of one: give a model tools, let it decide when to call them, execute the calls safely in your code, and loop until it has an answer. The result is a working Store Assistant that checks inventory and order status, in Scala 3 and Python.

The protocol in four messages

The wire format

{
  "model": "<model>",
  "messages": [
    { "role": "system", "content": "You help store staff. Use tools for stock and orders; never guess numbers." },
    { "role": "user",   "content": "Do we have tablets at Central?" }
  ],
  "tools": [
    { "type": "function",
      "function": {
        "name": "check_inventory",
        "description": "Get units available for one SKU at one store. Use for any stock/availability question.",
        "parameters": {
          "type": "object", "additionalProperties": false,
          "required": ["sku", "storeId"],
          "properties": {
            "sku":     { "type": "string", "enum": ["HEADPHONES", "PHONE_CASE", "CHARGER", "TABLET"] },
            "storeId": { "type": "string", "enum": ["RIVERSIDE", "CENTRAL", "AIRPORT"] }
          } } } }
  ],
  "temperature": 0
}

{ "choices": [ { "finish_reason": "tool_calls", "message": {
    "role": "assistant", "content": null,
    "tool_calls": [ { "id": "call_7f3", "type": "function",
      "function": { "name": "check_inventory", "arguments": "{\"sku\":\"TABLET\",\"storeId\":\"CENTRAL\"}" } } ]
} } ] }

{ "role": "tool", "tool_call_id": "call_7f3", "content": "{\"sku\":\"TABLET\",\"storeId\":\"CENTRAL\",\"available\":5}" }

Designing good tools

Scala 3: the Store Assistant

This single file contains tools with fake store data, a bounded agent loop, a scripted model so you can run it offline, and a live model that calls any OpenAI-compatible endpoint.

//> using scala 3
//> using dep io.circe::circe-core:0.14.15
//> using dep io.circe::circe-parser:0.14.15
import io.circe.*
import io.circe.parser.parse
import io.circe.syntax.*
import java.net.URI
import java.net.http.{HttpClient, HttpRequest, HttpResponse}

// ---------- Wire types ----------
final case class FunctionCall(name: String, arguments: String) derives Codec.AsObject
final case class ToolCall(id: String, `type`: String, function: FunctionCall) derives Codec.AsObject
final case class Msg(
    role: String,
    content: Option[String] = None,
    tool_calls: Option[List[ToolCall]] = None,
    tool_call_id: Option[String] = None
) derives Codec.AsObject

// ---------- Tools ----------
final case class Tool(name: String, description: String, parameters: Json, run: JsonObject => Either[String, Json])

def toolsJson(tools: List[Tool]): Json = tools.map { t =>
  Json.obj("type" -> "function".asJson,
    "function" -> Json.obj("name" -> t.name.asJson, "description" -> t.description.asJson, "parameters" -> t.parameters))
}.asJson

def str(args: JsonObject, key: String): Either[String, String] =
  args(key).flatMap(_.asString).toRight(s"missing string argument '$key'")

def enumSchema(values: Iterable[String]): Json = Json.obj("type" -> "string".asJson, "enum" -> values.asJson)
def objectSchema(props: (String, Json)*): Json = Json.obj(
  "type" -> "object".asJson, "additionalProperties" -> false.asJson,
  "required" -> props.map(_._1).asJson, "properties" -> Json.obj(props*))

// ---------- Fake store data (replace with real services) ----------
val Skus   = List("HEADPHONES", "PHONE_CASE", "CHARGER", "TABLET")
val Stores = List("RIVERSIDE", "CENTRAL", "AIRPORT")
val stock: Map[(String, String), Int] = Map(
  ("HEADPHONES", "RIVERSIDE") -> 3, ("HEADPHONES", "CENTRAL") -> 0, ("HEADPHONES", "AIRPORT") -> 7,
  ("TABLET", "CENTRAL") -> 5, ("CHARGER", "RIVERSIDE") -> 40
).withDefaultValue(0)
val orders: Map[String, Json] = Map(
  "ORD-482913" -> Json.obj("status" -> "READY_FOR_PICKUP".asJson, "storeId" -> "RIVERSIDE".asJson, "shelf" -> "B4".asJson),
  "ORD-500122" -> Json.obj("status" -> "IN_TRANSIT".asJson, "estimatedReady" -> "2026-10-10".asJson)
)

val tools = List(
  Tool("check_inventory",
    "Units available for one SKU at one store. Use for any stock or availability question about a specific store.",
    objectSchema("sku" -> enumSchema(Skus), "storeId" -> enumSchema(Stores)),
    args => for sku <- str(args, "sku"); store <- str(args, "storeId")
            yield Json.obj("sku" -> sku.asJson, "storeId" -> store.asJson, "available" -> stock((sku, store)).asJson)),
  Tool("find_stores_with_stock",
    "List stores that have at least one unit of a SKU. Use when the customer's store is out of stock.",
    objectSchema("sku" -> enumSchema(Skus)),
    args => str(args, "sku").map(sku =>
      Stores.flatMap(s => Option.when(stock((sku, s)) > 0)(Json.obj("storeId" -> s.asJson, "available" -> stock((sku, s)).asJson))).asJson)),
  Tool("get_order_status",
    "Status of a customer order by order ID (format ORD-123456). Use for pickup or delivery questions.",
    objectSchema("orderId" -> Json.obj("type" -> "string".asJson, "description" -> "e.g. ORD-482913".asJson)),
    args => str(args, "orderId").flatMap(id => orders.get(id).toRight(s"order '$id' not found")))
)

// ---------- Agent loop ----------
type Model = (List[Msg], Json) => Either[String, Msg]

val SystemPrompt =
  """You help retail store staff. Use the tools for stock and order facts; never guess numbers or statuses.
    |If a tool returns an error, explain it briefly. Answer in at most 3 sentences.""".stripMargin

def runAgent(model: Model, tools: List[Tool], question: String, maxSteps: Int = 6): String =
  val byName = tools.map(t => t.name -> t).toMap

  def execute(call: ToolCall): Msg =
    val output = for
      tool <- byName.get(call.function.name).toRight(s"unknown tool '${call.function.name}'")
      args <- parse(call.function.arguments).flatMap(_.as[JsonObject]).left.map(e => s"invalid arguments: ${e.getMessage}")
      out  <- tool.run(args)
    yield out
    val json = output.fold(err => Json.obj("error" -> err.asJson), identity)
    println(s"  [tool] ${call.function.name}(${call.function.arguments}) -> ${json.noSpaces}")
    Msg("tool", content = Some(json.noSpaces), tool_call_id = Some(call.id))

  @annotation.tailrec
  def loop(history: List[Msg], step: Int): String =
    if step > maxSteps then "I couldn't finish this safely. Please check the store system or ask a colleague."
    else model(history, toolsJson(tools)) match
      case Left(err) => s"AI unavailable: $err"
      case Right(reply) => reply.tool_calls.filter(_.nonEmpty) match
        case None        => reply.content.getOrElse("")
        case Some(calls) => loop(history ++ (reply :: calls.map(execute)), step + 1)

  loop(List(Msg("system", Some(SystemPrompt)), Msg("user", Some(question))), 1)

// ---------- Models ----------
/** Offline stand-in that behaves like a real model would for one question. */
val scriptedModel: Model = (history, _) =>
  val toolResults = history.filter(_.role == "tool")
  if toolResults.isEmpty then Right(Msg("assistant", tool_calls = Some(List(
    ToolCall("call_1", "function", FunctionCall("check_inventory", """{"sku":"HEADPHONES","storeId":"CENTRAL"}""")),
    ToolCall("call_2", "function", FunctionCall("find_stores_with_stock", """{"sku":"HEADPHONES"}"""))
  ))))
  else Right(Msg("assistant", content = Some(
    "Central is out of headphones right now. Riverside has 3 and Airport has 7, so we can transfer one or send the customer to Riverside.")))

/** Any OpenAI-compatible endpoint (local Ollama by default). */
def liveModel(base: String, model: String, key: Option[String]): Model = (history, toolsSpec) =>
  val body = Json.obj("model" -> model.asJson, "temperature" -> 0.asJson,
    "messages" -> history.asJson.deepDropNullValues, "tools" -> toolsSpec).noSpaces
  val b = HttpRequest.newBuilder(URI.create(s"$base/chat/completions"))
    .header("Content-Type", "application/json").POST(HttpRequest.BodyPublishers.ofString(body))
  key.foreach(k => b.header("Authorization", s"Bearer $k"))
  val resp = HttpClient.newHttpClient().send(b.build(), HttpResponse.BodyHandlers.ofString())
  if resp.statusCode() != 200 then Left(s"HTTP ${resp.statusCode()}: ${resp.body().take(300)}")
  else parse(resp.body()).flatMap(_.hcursor.downField("choices").downN(0).downField("message").as[Msg]).left.map(_.getMessage)

@main def storeAssistant(args: String*): Unit =
  val live     = args.headOption.contains("--live")
  val question = args.dropWhile(_ == "--live").headOption.getOrElse("A customer at Central wants headphones. Can we help?")
  val model =
    if live then liveModel(sys.env.getOrElse("LLM_BASE_URL", "http://localhost:11434/v1"),
                           sys.env.getOrElse("LLM_MODEL", "llama3.2"), sys.env.get("LLM_API_KEY"))
    else scriptedModel
  println(s"Q: $question")
  println(s"A: ${runAgent(model, tools, question)}")

Q: A customer at Central wants headphones. Can we help?
  [tool] check_inventory({"sku":"HEADPHONES","storeId":"CENTRAL"}) -> {"sku":"HEADPHONES","storeId":"CENTRAL","available":0}
  [tool] find_stores_with_stock({"sku":"HEADPHONES"}) -> [{"storeId":"RIVERSIDE","available":3},{"storeId":"AIRPORT","available":7}]
A: Central is out of headphones right now. Riverside has 3 and Airport has 7, so we can transfer one or send the customer to Riverside.

What makes this robust:

• Bounded loop (maxSteps) so a confused model can't loop forever or run up costs.
• Unknown tools, bad JSON and tool errors become tool results, not crashes, so the model can recover or explain.
• Enums in parameters stop the model inventing store IDs. Tool code still validates.
• The Model type is a function, so you can swap scripted, local and hosted models and unit-test the loop without a network.
• deepDropNullValues removes null fields that some servers reject.

Python version

import json
import os
from typing import Any, Callable

import httpx

STOCK = {("HEADPHONES", "RIVERSIDE"): 3, ("HEADPHONES", "AIRPORT"): 7, ("TABLET", "CENTRAL"): 5}
ORDERS = {"ORD-482913": {"status": "READY_FOR_PICKUP", "storeId": "RIVERSIDE", "shelf": "B4"}}
SKUS, STORES = ["HEADPHONES", "PHONE_CASE", "CHARGER", "TABLET"], ["RIVERSIDE", "CENTRAL", "AIRPORT"]


def check_inventory(sku: str, storeId: str) -> dict:
    return {"sku": sku, "storeId": storeId, "available": STOCK.get((sku, storeId), 0)}


def get_order_status(orderId: str) -> dict:
    return ORDERS.get(orderId) or {"error": f"order '{orderId}' not found"}


TOOLS: dict[str, tuple[Callable[..., dict], dict]] = {
    "check_inventory": (check_inventory, {
        "description": "Units available for one SKU at one store. Use for stock questions.",
        "parameters": {"type": "object", "additionalProperties": False, "required": ["sku", "storeId"],
                       "properties": {"sku": {"type": "string", "enum": SKUS},
                                      "storeId": {"type": "string", "enum": STORES}}}}),
    "get_order_status": (get_order_status, {
        "description": "Status of an order by ID (ORD-123456). Use for pickup questions.",
        "parameters": {"type": "object", "additionalProperties": False, "required": ["orderId"],
                       "properties": {"orderId": {"type": "string"}}}}),
}
TOOL_SPECS = [{"type": "function", "function": {"name": n, **spec}} for n, (_, spec) in TOOLS.items()]


def call_model(messages: list[dict]) -> dict:
    base = os.environ.get("LLM_BASE_URL", "http://localhost:11434/v1")
    key = os.environ.get("LLM_API_KEY")
    r = httpx.post(f"{base}/chat/completions", timeout=90,
                   headers={"Authorization": f"Bearer {key}"} if key else {},
                   json={"model": os.environ.get("LLM_MODEL", "llama3.2"), "temperature": 0,
                         "messages": messages, "tools": TOOL_SPECS})
    r.raise_for_status()
    return r.json()["choices"][0]["message"]


def execute(call: dict) -> dict:
    name = call["function"]["name"]
    try:
        fn, _ = TOOLS[name]
        result: Any = fn(**json.loads(call["function"]["arguments"] or "{}"))
    except KeyError:
        result = {"error": f"unknown tool '{name}'"}
    except (json.JSONDecodeError, TypeError) as e:
        result = {"error": f"invalid arguments: {e}"}
    print(f"  [tool] {name}({call['function']['arguments']}) -> {result}")
    return {"role": "tool", "tool_call_id": call["id"], "content": json.dumps(result)}


def run_agent(question: str, max_steps: int = 6) -> str:
    messages = [{"role": "system", "content": "You help retail store staff. Use tools; never guess stock or status."},
                {"role": "user", "content": question}]
    for _ in range(max_steps):
        reply = call_model(messages)
        calls = reply.get("tool_calls") or []
        if not calls:
            return reply.get("content") or ""
        messages.append(reply)
        messages.extend(execute(c) for c in calls)
    return "I couldn't finish this safely. Please check the store system."


if __name__ == "__main__":
    print(run_agent("Is order ORD-482913 ready, and where is it?"))

Write actions and human confirmation

Reading stock is harmless. Reserving stock, issuing a refund or messaging a customer changes the world. Common patterns:

• Propose, don't perform: the tool propose_transfer(sku, from, to, qty) returns a draft. A human clicks "Confirm" in your UI, and only then does your code execute it.
• Dry-run first: write tools take dryRun: true by default and return what would happen.
• Idempotency keys: retries of a write must not double-refund. Pass a key derived from the request.
• Audit log: record who asked, what the model proposed, and who approved.

Practical notes

• Parallel tool calls: a reply may contain several calls (as in the scripted example). Execute independent ones concurrently in production.
• tool_choice: most APIs let you force a specific tool, forbid tools, or let the model decide (default). Forcing a tool is a neat way to get structured output (page 02, level 4).
• Local models vary: smaller local models are less reliable at tool calling. Test with your real tools, and prefer models advertised for tool use.
• From functions to MCP: these same tools could be exposed by an MCP server (Level 1 page 09, built properly in Level 3). Then any MCP-capable app, such as your IDE agent, can use them too.