Human in the Loop

# Terminal 1: serve the example agent
motus serve start examples.serving.hitl_agent:agent --port 8000

# Terminal 2: chat with it
motus serve chat http://localhost:8000

from motus.tools import tool

@tool(requires_approval=True)
async def delete_file(path: str) -> str:
    """Delete a file at the given path."""
    import os
    os.remove(path)
    return f"Deleted {path}"

{
  "type": "tool_approval",
  "tool_name": "delete_file",
  "tool_args": { "path": "/tmp/old_logs.txt" }
}

{
  "interrupt_id": "<uuid from the interrupt>",
  "value": { "approved": true }
}

from motus.agent import ReActAgent
from motus.models import OpenRouterChatClient
from motus.tools import tool
from motus.tools.builtins.ask_user import ask_user_question

@tool
async def organize_files(directory: str, strategy: str) -> str:
    """Organize files in a directory using the chosen strategy."""
    return f"Organized {directory} by {strategy}"

agent = ReActAgent(
    client=OpenRouterChatClient(),
    model_name="anthropic/claude-sonnet-4",
    tools=[organize_files, ask_user_question],
)

{
  "type": "user_input",
  "questions": [
    {
      "question": "How would you like to organize the files?",
      "header": "Strategy",
      "multiSelect": false,
      "options": [
        { "label": "By date", "description": "Group files by year and month" },
        { "label": "By type", "description": "Group files by extension" },
        { "label": "By size", "description": "Move large files into a separate folder" }
      ]
    }
  ]
}

{
  "interrupt_id": "<uuid>",
  "value": {
    "answers": {
      "How would you like to organize the files?": "By date"
    }
  }
}

from motus.models import ChatMessage
from motus.serve.interrupt import interrupt

async def my_agent(message: ChatMessage, state: list[ChatMessage]):
    # An agent that takes (message, state) and returns (response, new_state)
    # is the plain callable agent shape that motus serve accepts.

    user_choice = await interrupt({
        "type": "color_picker",
        "prompt": "Pick a brand color for the export",
        "presets": ["#FF6B6B", "#4ECDC4", "#FFE66D"]
    })

    color = user_choice.get("hex", "#000000")
    response = ChatMessage.assistant_message(content=f"Using {color} for the export.")
    return response, state + [message, response]

idle ──POST /messages──▶ running ──┬──▶ idle           (turn completed)
                                    ├──▶ interrupted   (agent called interrupt)
                                    └──▶ error         (worker failed)

interrupted ──POST /resume──▶ running   (only after ALL pending interrupts are resolved)

import httpx

def run_turn(base_url: str, session_id: str, content: str) -> str:
    client = httpx.Client(timeout=60)

    # 1. Send the message
    client.post(f"{base_url}/sessions/{session_id}/messages",
                json={"role": "user", "content": content}).raise_for_status()

    # 2. Poll until idle or error, handling any interrupts that arrive
    while True:
        r = client.get(f"{base_url}/sessions/{session_id}",
                       params={"wait": "true", "timeout": "30"})
        data = r.json()
        status = data["status"]

        if status == "idle":
            return data["response"]["content"]
        if status == "error":
            raise RuntimeError(data["error"])
        if status == "interrupted":
            for intr in data["interrupts"]:
                value = handle_interrupt(intr)  # your UI logic
                client.post(f"{base_url}/sessions/{session_id}/resume",
                            json={"interrupt_id": intr["interrupt_id"], "value": value}
                ).raise_for_status()
            # loop and poll again, the worker is still running