Code Agent Anatomy (05): How Does the Model Know Which Tools Are Available? How Is Function Calling Implemented?

Follow a tool_call from start to finish: how tools register into the Registry, how the Registry generates function schemas to inform the model, how the Orchestrator batches execution after the model triggers tool_calls, the ToolExecutor's permission/optimistic-lock/circuit-breaker pipeline, why the ToolResult protocol separates internal and external representations, and how observation results are truncated before being written back to history.

·11 min read·AI Engineering
Code AgentFunction CallingTool SystemOpenAI ToolsLLM Agent

Starting with a Single tool_call

Suppose the model decides to read a file. Its response will contain:

{
  "tool_calls": [
    { "id": "call_abc", "name": "Read", "arguments": "{\"path\": \"tools/base.py\"}" }
  ]
}

Two questions arise:

  1. How does the model know the Read tool exists and what parameters it accepts?
  2. After the harness receives this response, how does it actually run the tool and feed the result back to the model?

To answer both questions, we need to start at the very beginning of the tool lifecycle — registration.

Step 1: Tool Registration — Python Classes Enter the Registry

All tools are registered into ToolRegistry when the agent starts up. The Registry maintains two internal tables, corresponding to two registration approaches:

# Approach 1: Tool object (recommended) — inherits the Tool base class, with fully structured parameter definitions
registry.register_tool(ReadFileTool(name="Read", description="...", project_root=...))
# Stored in self._tools: dict[str, Tool]
 
# Approach 2: Function registration (quick) — only needs a name, description, and a callable
registry.register_function("my_func", description="...", func=my_func)
# Stored in self._functions: dict[str, dict]

The key difference between the two approaches lies in parameter definitions: a Tool object returns a structured parameter list via get_parameters() (each parameter has a name, type, description, and required flag); function registration has no parameter definitions, so when generating the schema later, it can only fall back to a single input string.

Once registration is complete, the Registry becomes the "master catalog" of the tool system. Every subsequent LLM request fetches its schema from here.

Step 2: Before the API Request — the Registry Generates a Schema to Inform the Model

The model doesn't know about tools by default. At the start of each ReAct step, runtime/loop.py first retrieves the tools schema from the Registry, then sends it along with the request to the model:

# runtime/loop.py — _prepare_step_context() (loop.py:1007)
tools_schema = host._get_openai_tools_for_current_mode()
# Equivalent to self.tool_registry.get_openai_tools()
 
# runtime/loop.py — _react_loop() (loop.py:374)
raw_response = host.llm.invoke_raw(messages, tools=tools_schema, tool_choice=tool_choice)

get_openai_tools() converts both registration tables into OpenAI function schemas, so it has two for-loops, and the generated schema structures differ accordingly:

# tools/registry.py — get_openai_tools()
def get_openai_tools(self) -> list[dict]:
    tools = []
 
    # Loop 1: Process Tool objects (self._tools)
    # tool.get_parameters() returns a structured parameter list; _parameters_to_schema converts it to full JSON Schema
    for tool in sorted(self._tools.values(), key=lambda t: t.name):
        if not self._circuit_breaker.is_available(tool.name):
            continue  # Circuit-broken tools are not exposed to the model (covered in Step 3)
        tools.append({
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "parameters": self._parameters_to_schema(tool.get_parameters()),
                # Example result: {"path": string(required), "offset": integer, "limit": integer}
            },
        })
 
    # Loop 2: Process functions (self._functions)
    # Functions have no parameter definitions at registration time; the schema is hardcoded to a single input string
    for name, info in sorted(self._functions.items(), key=lambda item: item[0]):
        if not self._circuit_breaker.is_available(name):
            continue
        tools.append({
            "type": "function",
            "function": {
                "name": name,
                "description": info.get("description", ""),
                "parameters": {
                    "type": "object",
                    "properties": {"input": {"type": "string", "description": "raw input string"}},
                    "required": ["input"],
                    "additionalProperties": False,
                },
                # Regardless of the function's actual signature, the model always sees only one parameter: input
            },
        })
 
    return tools

The difference between the two registration types from the model's perspective:

Tool Object RegistrationFunction Registration
Parameters visible to modelComplete (name/type/description/required)Fixed: only input: string
Best suited forMultiple parameters, validation logicQuick integration, single input

The outcome of this step: the model receives a complete list of function schemas, knows which tools exist and what parameters each accepts, and can then decide which to call and what values to provide.

Step 3: The Model Returns tool_calls — the Orchestrator Takes Over

When the model decides to call a tool, its response includes tool_calls. Within a single ReAct loop step, the model may request multiple tools simultaneously (e.g., reading three files in parallel).

The entry point is ToolOrchestrator.run(), which handles two concerns: which calls can run concurrently, and how large an output is too large.

Parsing Arguments: plan_tool_calls

The first task is to parse the raw tool_calls list into an executable plan:

# tools/orchestrator.py — plan_tool_calls()
def plan_tool_calls(self, tool_calls: list[dict]) -> list[ToolCallPlan]:
    plans = []
    for call in tool_calls:
        tool_name = call.get("name") or "unknown_tool"
        tool_call_id = call.get("id") or f"call_{uuid.uuid4().hex}"
        raw_args = call.get("arguments") or {}
        # arguments may be a JSON string or already a dict; parse_tool_input handles both
        parsed_input, parse_error = parse_tool_input(raw_args)
        plans.append(ToolCallPlan(
            tool_name=tool_name,
            tool_call_id=tool_call_id,
            parsed_input=parsed_input if isinstance(parsed_input, dict) else {},
            parse_error=parse_error,          # Non-None on parse failure; short-circuits later
            concurrency_safe=self.is_concurrency_safe(tool_name, parse_error),
        ))
    return plans

Batching: partition_tool_calls

After parsing, plans are grouped into batches based on concurrency safety:

SAFE_TOOL_NAMES   = {"Read", "Grep", "Glob"}       # No side effects, safe to parallelize
UNSAFE_TOOL_NAMES = {"Edit", "Bash", "Task", ...}  # Side effects, must run serially
 
def partition_tool_calls(self, plans) -> list[ToolBatch]:
    batches = []
    for plan in plans:
        # Current plan is safe AND the last batch is also safe → merge into the same concurrent batch
        if batches and plan.concurrency_safe and batches[-1].concurrency_safe:
            batches[-1].calls.append(plan)
            continue
        # Otherwise, start a new batch
        batches.append(ToolBatch(concurrency_safe=plan.concurrency_safe, calls=[plan]))
    return batches

Example: if the model simultaneously requests [Read, Read, Edit, Grep], the batching result is:

[concurrent batch(Read, Read)]  →  [serial batch(Edit)]  →  [concurrent batch(Grep)]

Concurrent batches run in parallel via ThreadPoolExecutor; serial batches run sequentially one by one. Regardless of execution order, the order written back to history always matches the model's request order (concurrent results are reordered by offset).

Step 4: Each Tool's Execution Pipeline — ToolExecutor

Each plan is ultimately handed to ToolExecutor.execute(). This is where tools actually run, passing through four gates in sequence:

# tools/executor.py — execution pipeline
def execute(self, name: str, input_text: Any) -> ToolResult:
    parameters = self.registry.prepare_parameters(input_text)
 
    # Gate 1: Permission check
    if (denied := self._decide_permission(name, parameters)):
        return denied
 
    # Gate 2: Optimistic lock injection (Edit only)
    if name == "Edit":
        parameters = self.registry.inject_optimistic_lock_params(name, parameters)
 
    # Gate 3: Circuit breaker check
    if not self.registry.is_available(name):
        return self.registry.create_circuit_open_result(name, parameters)
 
    # Gate 4: Actual execution
    result = tool.run(parameters)
 
    # Post-execution: update circuit breaker; cache optimistic lock metadata for Read results
    self.registry.record_execution_result(name, result)
    if name == "Read":
        self.registry.cache_read_result(result, parameters)
 
    return result

Gate 1: Permission Check

RiskClassifier assigns one of three decisions — ALLOW / DENY / ASK — based on tool type and command content:

  • Read / Grep / Glob: ALLOW immediately (read-only, no side effects)
  • Edit: checks runtime_mode; DENY inside read-only sub-agents
  • Bash: regex matching — sudo, rm, git reset, nested shells get DENY; mv, pip install, etc. require ASK; low-risk read operations get ALLOW
  • Unknown tools: DENY by default (fail-closed policy)

Gate 2: Optimistic Lock Injection (Edit Only)

The typical agent pattern is Read then Edit. If someone modifies a file between the two calls, silently overwriting it would cause data loss.

The framework's solution: after a successful Read, cache the file's mtime and size; before an Edit executes, automatically inject these as expected values:

# After Read succeeds → cache metadata (tools/registry.py)
meta = {"file_mtime_ms": stats["file_mtime_ms"], "file_size_bytes": stats["file_size_bytes"]}
self._read_cache[path_resolved] = meta
 
# Before Edit → auto-inject from cache (tools/registry.py)
def _inject_optimistic_lock_params(self, tool_name, parameters):
    if "expected_mtime_ms" in parameters:
        return parameters       # Model already provided values; don't overwrite
    meta = self._read_cache.get(parameters.get("path"))
    if meta:
        parameters["expected_mtime_ms"] = meta["file_mtime_ms"]
        parameters["expected_size_bytes"] = meta["file_size_bytes"]
    return parameters

Inside the Edit tool, the expected values are compared against the file's current state. On conflict, a CONFLICT error is returned instead of silently overwriting.

Gate 3: Circuit Breaker Check

After N consecutive failures (default 3), a tool enters the OPEN state and rejects all calls for the duration of recovery_timeout (default 300 seconds):

CLOSED (normal) → N consecutive failures → OPEN (disabled)
                                             ↓ cooldown period ends
                                         HALF_OPEN (allow one attempt)
                                             ↓ success → CLOSED / failure → OPEN (timer resets)

While OPEN, the tool also disappears from get_openai_tools() output in Step 2 — it's not just rejected at execution time, the schema is no longer exposed to the model at all, so the model won't attempt to call it in the next step.

Step 5: What tool.run() Returns — the ToolResult Protocol

After passing all three gates, tool.run(parameters) is called. Every tool must return a ToolResult; returning a bare string is not allowed:

# tools/base.py
@dataclass(frozen=True)  # Immutable: cannot be accidentally modified as it flows through the pipeline
class ToolResult:
    status: ToolStatus       # success / partial / error
    text: str                # Summary for the LLM to read
    data: Dict[str, Any]     # Core payload (never None)
    error_code: ...          # Only meaningful when status is error
    stats: Dict[str, Any]    # Metrics such as time_ms
    context: Dict[str, Any]  # Context such as cwd and params_input

ToolResult is an internal object that flows between pipeline stages as a Python object without serialization. It is only converted to a JSON string at the very end, when written to history:

# tools/base.py — serialized only at the model boundary
def tool_result_payload(result: ToolResult) -> dict:
    payload = {"status": ..., "data": ..., "text": ..., "stats": ..., "context": ...}
    # The error field only appears when status=error, preventing the model from misreading successful results
    if result.status is ToolStatus.ERROR:
        payload["error"] = {"code": result.error_code.value, "message": ...}
    return payload

The reason for separating internal and external representations: if the pipeline serialized to a string mid-flow, each subsequent stage would need to re-parse the JSON, and strings can't be type-checked. ToolResult is a structured object; every pipeline stage can access fields by name and branch on status.

Step 6: Results Return to the Orchestrator — Budget Truncation Before Writing to History

After _execute_one() receives the ToolResult, the Orchestrator wraps it into a ToolObservation, which performs serialization automatically at construction time:

@dataclass(frozen=True)
class ToolObservation:
    result: ToolResult      # For internal pipeline use
    raw_result: ToolResult  # Original result before truncation (for debugging)
    observation: str        # Serialized JSON, written directly to history
 
    def __post_init__(self):
        object.__setattr__(self, "observation", serialize_tool_result(self.result))

After all batches complete, two layers of budget control are applied to prevent tool output from overwhelming the context window:

Layer 1 (per tool):  result > 50KB → force truncate; full content spills to disk; result includes the file path
Layer 2 (total):     sum of all results > 200KB → truncate from the largest un-truncated item until total is within budget

Finally, the observation string is written to history keyed by tool_call_id:

{ "role": "tool", "tool_call_id": "call_abc", "content": "{...ToolResult JSON...}" }

The model reads this record in the next step and continues reasoning.

End-to-End Recap

① At startup
   register_tool / register_function → self._tools / self._functions
 
② Before each ReAct step's LLM request
   get_openai_tools() → tools schema → llm.invoke_raw(tools=...) → model knows which tools exist
 
③ Model returns tool_calls
   ToolOrchestrator.run()
     plan_tool_calls()      parse arguments, annotate concurrency safety
     partition_tool_calls() batch (read-only → concurrent, write → serial)
 
④ Execute each batch
   _execute_plan() → ToolExecutor.execute()
     Permission check (ALLOW/DENY/ASK)
     Optimistic lock injection (Edit: auto-inject expected_mtime_ms)
     Circuit breaker check (OPEN state rejects immediately)
     tool.run(parameters) → ToolResult
 
⑤ Results flow back
   ToolResult → ToolObservation (serialized at construction)
   Two-layer budget truncation (50KB per tool + 200KB total)
   → history tool message (role=tool, content=JSON)

Registration determines that a tool exists; schema generation determines what the model can see; the model triggers tool_calls; the Orchestrator batches them; the Executor enforces the gates; ToolResult flows internally and is only serialized at the boundary; finally written back to history to complete one action-observation cycle in ReAct.

About the Source Code for This Series

All analysis in this series is based on the open-source project MyCodeAgent.

The source code has been annotated at key locations in the order covered by this series — you can follow along with the code as you read, or clone it and run, modify, and extend it yourself to build your own agent.

git clone https://github.com/chendongqi/MyCodeAgent
cd MyCodeAgent
cp .env.example .env   # Fill in your LLM API key
uv sync
uv run python main.py

Visit PrimeSkills — a curated marketplace of AI Agent workflows and skills, all validated against real enterprise workloads. No hype, just things that actually work.

For more practical insights and interesting products, visit my personal homepage

← Back to Blog