CourseModel Context Protocol · Module 11: Capstone · part 77 of 83
Part 77 · Module 11: Capstone

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_tool handles the model's requests in order of strictness. The host-side read_resource tool reads a notes:// URI. A tool blocked by pinning, or an unknown tool, gets an explanation instead of a call. Any tool that needs_approval() goes through self.approve first; a "no" becomes a tool result telling the model not to retry. Every real call has a timeout. In ask, the loop runs at most max_iterations model turns; a call whose name and arguments repeat more than max_repeat_calls times stops the loop; exceptions become text for the model rather than crashing the host. The log line records sorted(call.arguments), which is the argument keys only: a create_note body or a search query could contain note text, so values never reach the log.
  • Comes out: a HostAnswer with text, stop_reason (answered, max_iterations, or repeated_call), and one ToolCallRecord per call. The runs below show all three stop reasons.

The rest of this course is yours to keep

This course is bought on its own, once, and stays readable afterwards, including the parts added to it later.