Topic 4: The host and its CLI
14 min read·22 Sept 2026
The loop, the approval gate, and the log line
The host is the program that owns the conversation with the model and decides what the model may do. Module 6 built it and Module 8 added pinning. These two methods are where three capstone requirements live: the loop guards, the approval step, and the logging rule.
python
# Excerpt from notes_assistant/host.py, lines 172 to 226, dedented (methods of class Host)
async def _run_tool(self, name: str, arguments: dict[str, Any]) -> tuple[str, bool, str]:
"""Execute one model-requested call. Returns (text for the model, ok, note)."""
if name == READ_RESOURCE_TOOL:
uri = str(arguments.get("uri", ""))
with anyio.fail_after(self.config.tool_timeout_seconds):
resource = await self._client_for_uri(uri).read_resource(uri)
texts = [c.text for c in resource.contents if hasattr(c, "text")]
return "\n".join(texts), True, ""
if name in self.blocked:
return f"Tool {name!r} is disabled: {self.blocked[name]}.", False, "pin_mismatch"
if name not in self.tools:
return f"Unknown tool {name!r}. Available: {sorted(self.tools)}", False, "unknown_tool"
server_name, tool = self.tools[name]
if needs_approval(tool) and not await self._approved(name, arguments):
return "The user declined this action. Do not retry it; tell the user it was not done.", False, "declined"
with anyio.fail_after(self.config.tool_timeout_seconds):
result = await self.clients[server_name].call_tool(tool.name, arguments)
return result_text(result), not result.is_error, "tool_error" if result.is_error else ""
async def ask(self, question: str) -> HostAnswer:
"""Run the agent loop for one question."""
specs = await self.load_tools()
messages: list[dict[str, Any]] = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": question},
]
calls: list[ToolCallRecord] = []
seen: dict[str, int] = {}
for _ in range(self.config.max_iterations):
reply = self.chat_fn(messages, specs)
if not reply.tool_calls:
return HostAnswer(reply.content or "", "answered", calls)
messages.append(reply.as_message())
for call in reply.tool_calls:
key = call.name + json.dumps(call.arguments, sort_keys=True)
seen[key] = seen.get(key, 0) + 1
if seen[key] > self.config.max_repeat_calls:
logger.warning("stopping: repeated call tool=%s", call.name)
return HostAnswer("I stopped because the same tool call kept repeating.", "repeated_call", calls)
started = time.perf_counter()
try:
text, ok, note = await self._run_tool(call.name, call.arguments)
except TimeoutError:
text, ok, note = "The tool timed out. Try a narrower request.", False, "timeout"
except Exception as exc: # a failing server must not crash the host
text, ok, note = f"The tool failed: {type(exc).__name__}.", False, "exception"
duration_ms = int((time.perf_counter() - started) * 1000)
calls.append(ToolCallRecord(call.name, call.arguments, ok, duration_ms, note))
# Log names and argument keys only: argument values may contain note text.
logger.info(
"tool_call name=%s arg_keys=%s ok=%s ms=%d note=%s",
call.name, sorted(call.arguments), ok, duration_ms, note,
)
messages.append({"role": "tool", "tool_call_id": call.id, "content": text})
return HostAnswer("I stopped after reaching the step limit.", "max_iterations", calls)Code explained
- In simple words: the host asks the model what to do, checks every request against its rules, runs the allowed ones, and writes one short log line per call.
- What happens:
_run_toolhandles the model's requests in order of strictness. The host-sideread_resourcetool reads anotes://URI. A tool blocked by pinning, or an unknown tool, gets an explanation instead of a call. Any tool thatneeds_approval()goes throughself.approvefirst; a "no" becomes a tool result telling the model not to retry. Every real call has a timeout. Inask, the loop runs at mostmax_iterationsmodel turns; a call whose name and arguments repeat more thanmax_repeat_callstimes stops the loop; exceptions become text for the model rather than crashing the host. The log line recordssorted(call.arguments), which is the argument keys only: acreate_notebody or a search query could contain note text, so values never reach the log. - Comes out: a
HostAnswerwithtext,stop_reason(answered,max_iterations, orrepeated_call), and oneToolCallRecordper call. The runs below show all three stop reasons.