How to Build a Cursor-Like AI Coding Agent — Complete Guide

livecode-ai/                    ~/livecode/  (runtime)
├── src/livecode/
│   ├── server.py               ├── settings.json
│   ├── harness/turn.py         └── projects/{slug}/
│   ├── tools.py                    ├── sessions/{id}/chat_history.jsonl
│   ├── runtime.py                  ├── index/workspace.json
│   ├── session.py                  ├── symbols/symbols.json
│   └── memory/                     └── memory/index.sqlite
├── templates/index.html
└── static/js/bundle.js

@app.post("/livecode-agent")
def agent_turn():
    body = request.get_json()
    question = body["question"]
    project_path = body["project_path"]
    session_id = body.get("session_id") or uuid.uuid4().hex
    mode = body.get("mode", "agent")

    def generate():
        bridge = SSEProgressBridge(socketio=socketio, room=socket_id)
        chunk_queue = queue.Queue()
        turn_thread = threading.Thread(
            target=lambda: run_turn_in_queue(run_livecode_turn(...), chunk_queue),
            daemon=True,
        )
        turn_thread.start()

        while not turn_finished:
            yield from bridge.drain_as_sse()
            try:
                kind, payload = chunk_queue.get(timeout=0.05)
            except queue.Empty:
                continue
            if payload.get("done"):
                append_turn_messages(project_path, session_id, payload["turn_messages"])
                yield f"data: {json.dumps(payload)}\n\n"
                break
            yield payload

    return Response(
        stream_with_context(generate()),
        mimetype="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

def run_livecode_turn(project_path, question, *, mode="agent", session_id, ...):
    classification = classify_turn(question)
    messages = build_base_messages(project_path, question, session_id)
    tools = filter_tools_for_mode(get_livecode_tools(), mode)

    for iteration in range(1, LIVECODE_MAX_ITERATIONS + 1):
        if stationarity_detected(tool_history):
            messages.append(nudge("same tool+args repeated — try a different approach"))
        if search_scatter_detected(tool_history):
            messages.append(nudge("broaden grep — use glob or alternation"))
        if exploration_streak_without_edit(tool_history):
            messages.append(nudge("you have enough context — edit or finish"))

        messages = compact_stale_tool_messages(messages, keep_last=6)

        if interjection := drain_interjection(session_id):
            messages.append(format_interjection(interjection))

        iter_tools = tools
        if iteration > LIVECODE_MAX_ITERATIONS - CLOSURE_ITERATIONS:
            iter_tools = attempt_completion_only_tools(tools)

        response = call_with_tools(
            model=pick_auto_model(classification),
            messages=messages,
            tools=iter_tools,
            on_thought_delta=lambda d: emit("agent_thinking_delta", d),
        )

        if not response.tool_calls:
            yield sse_done(answer=response.content)
            return

        for result in execute_tool_calls_batch(response.tool_calls):
            messages.append(assistant_tool_message(response))
            messages.append(tool_result_message(result))
            emit_progress(result)

            if result.tool == "attempt_completion":
                if goal_verifier_passed(question, tool_history):
                    yield sse_done(answer=result.summary)
                    return
                messages.append(nudge("user asked for code change but no edit succeeded"))

    yield sse_done(answer=summarize_exhaustion(messages))

LIVECODE_MAX_ITERATIONS = 100
LIVECODE_CONTEXT_WINDOW = 128_000
STATIONARITY_NUDGE_AFTER = 8
STATIONARITY_HARD_STOP = 16
CLOSURE_ITERATIONS = 2

def dispatch_tool(project_path, name, args, *, mode, session_id, ...):
    if mode_blocks_tool(mode, name):
        return {"error": mode_rejection_message(mode, name)}

    if name == "run_command" and requires_permission(args["command"]):
        request_id = create_permission_request(session_id, name, args)
        emit_permission_prompt(request_id, args)
        approved = wait_for_permission(request_id)
        if not approved:
            return {"error": "User denied command"}

    if name == "grep_repo":
        return repo_grep_fn(project_path, pattern=args["pattern"], ...)
    if name == "edit_file":
        return apply_search_replace(
            project_path, args["file_path"],
            old=args["old_string"], new=args["new_string"],
        )

LIVECODE_AUTO_COMPACT_RATIO = 0.85
LIVECODE_IN_TURN_COMPACT_RATIO = 0.70
LIVECODE_INTER_COMPACT_RATIO = 0.65
LIVECODE_KEEP_RECENT_TOOL_MSGS = 6

const resp = await fetch("/livecode-agent", {
  method: "POST",
  headers: { "Content-Type": "application/json", Accept: "text/event-stream" },
  body: JSON.stringify({ project_path, question, session_id, mode }),
});

const reader = resp.body.getReader();
const decoder = new TextDecoder();
let buffer = "";

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  buffer += decoder.decode(value, { stream: true });

  for (const line of buffer.split("\n")) {
    if (!line.startsWith("data: ")) continue;
    const payload = JSON.parse(line.slice(6));
    if (payload.progress) handleProgress(payload.progress);
    if (payload.command_stream) appendTerminalOutput(payload.command_stream);
    if (payload.done) finalizeTurn(payload);
  }
}

def create_permission_request(session_id, tool_name, tool_args, timeout_s=120):
    request_id = f"perm_{uuid.uuid4().hex[:16]}"
    event = threading.Event()
    _PERMISSIONS[request_id] = {
        "session_id": session_id,
        "tool_name": tool_name,
        "tool_args": tool_args,
        "event": event,
        "approved": None,
    }
    return request_id

def wait_for_permission(request_id) -> bool | None:
    entry = _PERMISSIONS[request_id]
    if not entry["event"].wait(timeout=entry["timeout_s"]):
        return None
    return bool(entry["approved"])

LIVECODE_MODES = ("agent", "plan", "ask")

def filter_tools_for_mode(tools, mode):
    if mode == "agent":
        return tools
    allowed = READ_ONLY_TOOLS | {"attempt_completion"}
    filtered = [t for t in tools if t["function"]["name"] in allowed]
    if mode == "plan":
        filtered.append(CREATE_PLAN_TOOL)
    return filtered

def run_subagent_turn(project_path, goal, parent_session_id, read_only=True, max_iterations=5):
    child_session = f"{parent_session_id}_sub_{uuid.uuid4().hex[:8]}"
    question = (
        f"[Subagent — read_only={read_only}]\n{goal}\n"
        "Call attempt_completion when done."
    )

    for chunk in run_livecode_turn(
        project_path, question,
        session_id=child_session,
        max_iterations=max_iterations,
        mode="ask" if read_only else "agent",
    ):
        collect_answer_from_sse_chunk(chunk)

    return {"success": True, "result": answer, "child_session_id": child_session}

{"role":"user","content":"Add health check endpoint","ts":1725000000}
{"role":"assistant","content":null,"tool_calls":[{"id":"call_1","function":{"name":"grep_repo","arguments":"{\"pattern\":\"router\"}"}}]}
{"role":"tool","tool_call_id":"call_1","content":"{\"matches\":[{\"file\":\"app.py\",\"line\":12}]}"}

class IntelligentClassification(TypedDict):
    goal_kind: str          # code_change, analysis, research, meta
    edit_scope: str         # none, single_line, single_file, multi_file, bulk
    needs_flagship_model: bool
    complexity: str         # simple, medium, complex
    expects_multi_step: bool

RULES_FILES = ["AGENTS.md", "CLAUDE.md", ".cursor/rules", ".livecode/rules"]

def discover_project_rules(project_path):
    for directory in path_chain(git_root, project_path):
        for rules_file in RULES_FILES:
            if exists(join(directory, rules_file)):
                found.append(read(candidate))
    return wrap_in_system_reminder(found)