Add v1.1.0 with progress streaming, timeouts, and git validation.

Surface Cursor run progress to stderr and response payloads, enforce
CURSOR_TIMEOUT_SECONDS, validate git repos before delegate, and fix
resume session cwd persistence. Includes docs, tests, and packaging.

Co-authored-by: Cursor <cursoragent@cursor.com>

src/cursor_mcp/__init__.py

@@ -0,0 +1 @@
+__version__ = "1.1.0"

src/cursor_mcp/audit.py

@@ -0,0 +1,22 @@
+"""Audit log for Cursor delegations."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from cursor_mcp.config import settings
+
+
+def audit_log(event: str, payload: dict[str, Any]) -> None:
+    path = Path(settings.audit_log_path).expanduser()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    record = {
+        "ts": datetime.now(timezone.utc).isoformat(),
+        "event": event,
+        **payload,
+    }
+    with path.open("a", encoding="utf-8") as fh:
+        fh.write(json.dumps(record, default=str) + "\n")

src/cursor_mcp/client.py

@@ -0,0 +1,310 @@
+"""Cursor SDK wrapper used by MCP tools."""
+
+from __future__ import annotations
+
+import os
+import time
+from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
+from pathlib import Path
+from typing import Any
+
+from cursor_sdk import Agent, AgentOptions, CursorAgentError, LocalAgentOptions, Run, SendOptions
+
+from cursor_mcp.audit import audit_log
+from cursor_mcp.config import settings
+from cursor_mcp.errors import CursorMCPError
+from cursor_mcp.progress import ProgressCollector
+from cursor_mcp.repo import validate_git_repo
+from cursor_mcp.sessions import get_session, list_sessions, save_session
+
+
+def _require_api_key() -> str:
+    key = settings.api_key or os.environ.get("CURSOR_API_KEY", "")
+    if not key:
+        raise CursorMCPError(
+            "CURSOR_API_KEY is not set. Add it to ~/.hermes/.env or export it in the MCP env."
+        )
+    return key
+
+
+def _resolve_cwd(cwd: str = "") -> Path:
+    raw = (cwd or settings.default_cwd).strip()
+    path = Path(raw).expanduser().resolve()
+    if not path.exists():
+        raise CursorMCPError(f"Working directory does not exist: {path}")
+    if not path.is_dir():
+        raise CursorMCPError(f"Working directory is not a folder: {path}")
+    return path
+
+
+def _options(*, cwd: Path, model: str = "") -> AgentOptions:
+    return AgentOptions(
+        api_key=_require_api_key(),
+        model=model or settings.default_model,
+        local=LocalAgentOptions(cwd=str(cwd)),
+    )
+
+
+def _send_options(*, model: str = "", collector: ProgressCollector | None = None) -> SendOptions:
+    return SendOptions(
+        model=model or None,
+        on_step=collector.on_step if collector else None,
+        on_delta=collector.on_delta if collector else None,
+    )
+
+
+def _format_run_result(
+    result: Any,
+    *,
+    agent_id: str | None = None,
+    progress: list[dict[str, Any]] | None = None,
+    repo: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    payload: dict[str, Any] = {
+        "status": str(getattr(result, "status", "unknown")),
+        "run_id": getattr(result, "id", None),
+        "agent_id": getattr(result, "agent_id", None) or agent_id,
+        "result": getattr(result, "result", "") or "",
+        "duration_ms": getattr(result, "duration_ms", None),
+        "model": getattr(result, "model", None),
+    }
+    git = getattr(result, "git", None)
+    if git is not None:
+        payload["git"] = {
+            "branch": getattr(git, "branch", None),
+            "commit": getattr(git, "commit", None),
+        }
+    elif repo is not None:
+        payload["git"] = {
+            "branch": repo.get("branch"),
+            "commit": repo.get("commit"),
+            "root": repo.get("root"),
+            "dirty": repo.get("dirty"),
+        }
+    if progress:
+        payload["progress"] = progress
+    return payload
+
+
+def _wait_run(run: Run, collector: ProgressCollector, timeout_seconds: int) -> Any:
+    if run._terminal_result is not None:
+        return run._terminal_result
+
+    deadline = time.monotonic() + timeout_seconds
+    run.on_did_change_status(collector.on_status)
+
+    for _ in run.events():
+        if time.monotonic() > deadline:
+            _cancel_run(run)
+            raise CursorMCPError(
+                f"Cursor run timed out after {timeout_seconds}s (run_id={run.id or 'pending'})"
+            )
+
+    if run._terminal_result is not None:
+        return run._terminal_result
+
+    remaining = max(1.0, deadline - time.monotonic())
+    return _wait_live_run(run, remaining, timeout_seconds)
+
+
+def _wait_live_run(run: Run, remaining_seconds: float, configured_timeout: int) -> Any:
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        future = executor.submit(run.client.wait_live_run, run.id)
+        try:
+            result = future.result(timeout=remaining_seconds)
+        except FuturesTimeoutError as exc:
+            _cancel_run(run)
+            raise CursorMCPError(
+                f"Cursor run timed out after {configured_timeout}s (run_id={run.id})"
+            ) from exc
+
+    for callback in run._apply_result(result):
+        callback()
+    run._terminal_result = result
+    return result
+
+
+def _cancel_run(run: Run) -> None:
+    try:
+        run.cancel()
+    except Exception:
+        pass
+    audit_log("run_cancelled", {"run_id": run.id, "agent_id": run.agent_id})
+
+
+def _execute_send(
+    *,
+    agent: Agent,
+    prompt: str,
+    cwd: Path,
+    model: str,
+    collector: ProgressCollector,
+) -> dict[str, Any]:
+    run = agent.send(prompt, _send_options(model=model, collector=collector))
+    result = _wait_run(run, collector, settings.timeout_seconds)
+    return _format_run_result(
+        result,
+        agent_id=getattr(agent, "id", None),
+        progress=collector.snapshot(),
+    )
+
+
+def delegate_once(
+    *,
+    prompt: str,
+    cwd: str = "",
+    model: str = "",
+    task_summary: str = "",
+) -> dict[str, Any]:
+    if not prompt.strip():
+        raise CursorMCPError("prompt is required")
+    workdir = _resolve_cwd(cwd)
+    repo = validate_git_repo(workdir, required=settings.require_git_repo)
+    collector = ProgressCollector()
+    audit_log(
+        "delegate_once_start",
+        {
+            "cwd": str(workdir),
+            "model": model or settings.default_model,
+            "summary": task_summary,
+            "timeout_seconds": settings.timeout_seconds,
+        },
+    )
+    agent = Agent.create(_options(cwd=workdir, model=model))
+    try:
+        payload = _execute_send(
+            agent=agent,
+            prompt=prompt,
+            cwd=workdir,
+            model=model,
+            collector=collector,
+        )
+    except CursorAgentError as exc:
+        audit_log("delegate_once_error", {"message": str(exc), "retryable": exc.is_retryable})
+        raise CursorMCPError(f"Cursor agent failed to start: {exc}") from exc
+    finally:
+        agent.close()
+
+    if payload.get("agent_id") and payload.get("run_id"):
+        save_session(
+            agent_id=str(payload["agent_id"]),
+            run_id=str(payload["run_id"]),
+            cwd=str(workdir),
+            task_summary=task_summary or prompt[:120],
+            status=str(payload["status"]),
+            result_preview=str(payload.get("result", "")),
+        )
+    audit_log("delegate_once_done", {"status": payload.get("status"), "run_id": payload.get("run_id")})
+    if repo is not None and "git" not in payload:
+        payload["git"] = repo
+    return payload
+
+
+def delegate_session(
+    *,
+    prompt: str,
+    cwd: str = "",
+    model: str = "",
+    task_summary: str = "",
+) -> dict[str, Any]:
+    """Create a durable Cursor agent, run first prompt, return agent_id for follow-ups."""
+    if not prompt.strip():
+        raise CursorMCPError("prompt is required")
+    workdir = _resolve_cwd(cwd)
+    repo = validate_git_repo(workdir, required=settings.require_git_repo)
+    collector = ProgressCollector()
+    audit_log(
+        "delegate_session_start",
+        {
+            "cwd": str(workdir),
+            "model": model or settings.default_model,
+            "summary": task_summary,
+            "timeout_seconds": settings.timeout_seconds,
+        },
+    )
+    try:
+        with Agent.create(
+            api_key=_require_api_key(),
+            model=model or settings.default_model,
+            local=LocalAgentOptions(cwd=str(workdir)),
+        ) as agent:
+            payload = _execute_send(
+                agent=agent,
+                prompt=prompt,
+                cwd=workdir,
+                model=model,
+                collector=collector,
+            )
+            agent_id = payload.get("agent_id")
+            if agent_id and payload.get("run_id"):
+                save_session(
+                    agent_id=str(agent_id),
+                    run_id=str(payload["run_id"]),
+                    cwd=str(workdir),
+                    task_summary=task_summary or prompt[:120],
+                    status=str(payload["status"]),
+                    result_preview=str(payload.get("result", "")),
+                )
+    except CursorAgentError as exc:
+        audit_log("delegate_session_error", {"message": str(exc), "retryable": exc.is_retryable})
+        raise CursorMCPError(f"Cursor agent failed: {exc}") from exc
+
+    audit_log("delegate_session_done", payload)
+    if repo is not None and "git" not in payload:
+        payload["git"] = repo
+    return payload
+
+
+def resume_agent(
+    *,
+    agent_id: str,
+    prompt: str,
+    model: str = "",
+) -> dict[str, Any]:
+    if not agent_id.strip():
+        raise CursorMCPError("agent_id is required")
+    if not prompt.strip():
+        raise CursorMCPError("prompt is required")
+
+    prior = get_session(agent_id)
+    resume_cwd = (prior or {}).get("cwd") or settings.default_cwd
+    collector = ProgressCollector()
+    audit_log(
+        "resume_start",
+        {"agent_id": agent_id, "cwd": resume_cwd, "timeout_seconds": settings.timeout_seconds},
+    )
+    try:
+        with Agent.resume(agent_id, AgentOptions(api_key=_require_api_key(), model=model or None)) as agent:
+            payload = _execute_send(
+                agent=agent,
+                prompt=prompt,
+                cwd=Path(resume_cwd),
+                model=model,
+                collector=collector,
+            )
+    except CursorAgentError as exc:
+        audit_log("resume_error", {"agent_id": agent_id, "message": str(exc)})
+        raise CursorMCPError(f"Cursor resume failed: {exc}") from exc
+
+    if payload.get("run_id"):
+        save_session(
+            agent_id=agent_id,
+            run_id=str(payload["run_id"]),
+            cwd=resume_cwd,
+            task_summary=prompt[:120],
+            status=str(payload.get("status", "")),
+            result_preview=str(payload.get("result", "")),
+        )
+    audit_log("resume_done", payload)
+    return payload
+
+
+def get_status() -> dict[str, Any]:
+    return {
+        "api_key_configured": bool(settings.api_key or os.environ.get("CURSOR_API_KEY")),
+        "default_model": settings.default_model,
+        "default_cwd": settings.default_cwd,
+        "timeout_seconds": settings.timeout_seconds,
+        "require_git_repo": settings.require_git_repo,
+        "recent_sessions": list_sessions(),
+    }

src/cursor_mcp/config.py

@@ -0,0 +1,22 @@
+"""Configuration for cursor-mcp."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_prefix="CURSOR_", extra="ignore")
+
+    api_key: str = ""
+    default_model: str = "composer-2.5"
+    default_cwd: str = str(Path.home() / "Projects")
+    timeout_seconds: int = 1800
+    require_git_repo: bool = True
+    sessions_path: str = str(Path.home() / ".local/share/cursor-mcp/sessions.json")
+    audit_log_path: str = str(Path.home() / ".local/share/cursor-mcp/audit.jsonl")
+
+
+settings = Settings()

src/cursor_mcp/errors.py

@@ -0,0 +1,5 @@
+"""Shared cursor-mcp exceptions."""
+
+
+class CursorMCPError(Exception):
+    pass

src/cursor_mcp/progress.py

@@ -0,0 +1,114 @@
+"""Collect and emit Cursor run progress for Hermes MCP consumers."""
+
+from __future__ import annotations
+
+import sys
+from datetime import datetime, timezone
+from typing import Any
+
+from cursor_sdk.types import (
+    ConversationStep,
+    InteractionUpdate,
+    StepCompletedUpdate,
+    StepStartedUpdate,
+    SummaryUpdate,
+    ToolCallCompletedUpdate,
+    ToolCallStartedUpdate,
+)
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _tool_label(tool_call: Any) -> str:
+    if not isinstance(tool_call, dict):
+        tool_call = getattr(tool_call, "__dict__", {}) or {}
+    if isinstance(tool_call, dict):
+        name = tool_call.get("name") or tool_call.get("toolName") or tool_call.get("type")
+        if name:
+            return str(name)
+    return "tool"
+
+
+class ProgressCollector:
+    """Record progress events and mirror key lines to stderr (Hermes mcp-stderr.log)."""
+
+    def __init__(self, *, emit_stderr: bool = True, max_events: int = 100) -> None:
+        self.emit_stderr = emit_stderr
+        self.max_events = max_events
+        self.events: list[dict[str, Any]] = []
+
+    def emit(self, kind: str, message: str, **extra: Any) -> None:
+        entry: dict[str, Any] = {
+            "ts": _utc_now(),
+            "kind": kind,
+            "message": message,
+            **extra,
+        }
+        self.events.append(entry)
+        if len(self.events) > self.max_events:
+            self.events = self.events[-self.max_events :]
+        if self.emit_stderr and message:
+            print(f"[cursor-mcp] {kind}: {message}", file=sys.stderr, flush=True)
+
+    def on_status(self, status: str) -> None:
+        self.emit("status", status, status=status)
+
+    def on_step(self, step: ConversationStep) -> None:
+        step_type = getattr(step, "type", "step")
+        if step_type == "toolCall":
+            message = step.message if isinstance(step.message, dict) else {}
+            label = _tool_label(message)
+            self.emit("step", f"tool call: {label}", step_type=step_type, tool=label)
+            return
+        if step_type == "assistantMessage":
+            text = getattr(getattr(step, "message", None), "text", "") or ""
+            preview = text.strip().replace("\n", " ")[:120]
+            if preview:
+                self.emit("step", f"assistant: {preview}", step_type=step_type)
+            return
+        if step_type == "thinkingMessage":
+            self.emit("step", "thinking", step_type=step_type)
+            return
+        if step_type == "shellConversationTurn":
+            self.emit("step", "shell command", step_type=step_type)
+            return
+        self.emit("step", step_type, step_type=step_type)
+
+    def on_delta(self, update: InteractionUpdate) -> None:
+        update_type = getattr(update, "type", None)
+        if update_type is None and isinstance(update, dict):
+            update_type = update.get("type")
+        if update_type == "tool-call-started":
+            tool = _tool_label(getattr(update, "tool_call", {}))
+            self.emit("delta", f"started {tool}", update_type=update_type, tool=tool)
+            return
+        if update_type == "tool-call-completed":
+            tool = _tool_label(getattr(update, "tool_call", {}))
+            self.emit("delta", f"completed {tool}", update_type=update_type, tool=tool)
+            return
+        if update_type == "summary":
+            summary = getattr(update, "summary", "") or ""
+            preview = summary.strip().replace("\n", " ")[:160]
+            if preview:
+                self.emit("delta", preview, update_type=update_type)
+            return
+        if update_type == "step-started":
+            step_id = getattr(update, "step_id", None)
+            self.emit("delta", f"step {step_id} started", update_type=update_type, step_id=step_id)
+            return
+        if update_type == "step-completed":
+            step_id = getattr(update, "step_id", None)
+            duration = getattr(update, "step_duration_ms", None)
+            self.emit(
+                "delta",
+                f"step {step_id} completed ({duration}ms)" if duration else f"step {step_id} completed",
+                update_type=update_type,
+                step_id=step_id,
+                duration_ms=duration,
+            )
+            return
+
+    def snapshot(self) -> list[dict[str, Any]]:
+        return list(self.events)

src/cursor_mcp/repo.py

@@ -0,0 +1,57 @@
+"""Optional git repository validation before delegation."""
+
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+from typing import Any
+
+from cursor_mcp.errors import CursorMCPError
+
+
+def validate_git_repo(workdir: Path, *, required: bool = True) -> dict[str, Any] | None:
+    """Verify ``workdir`` is inside a git work tree when ``required`` is true."""
+    if not required:
+        return None
+
+    inside = _run_git(workdir, ["rev-parse", "--is-inside-work-tree"])
+    if inside.stdout.strip() != "true":
+        raise CursorMCPError(
+            f"Working directory is not a git repository: {workdir}. "
+            "Initialize git or set CURSOR_REQUIRE_GIT_REPO=false for scratch work."
+        )
+
+    root = _run_git(workdir, ["rev-parse", "--show-toplevel"]).stdout.strip()
+    branch = _run_git(workdir, ["rev-parse", "--abbrev-ref", "HEAD"]).stdout.strip()
+    commit = _run_git(workdir, ["rev-parse", "--short", "HEAD"]).stdout.strip()
+    dirty = _run_git(workdir, ["status", "--porcelain"]).stdout.strip()
+
+    return {
+        "root": root,
+        "branch": branch,
+        "commit": commit,
+        "dirty": bool(dirty),
+    }
+
+
+def _run_git(workdir: Path, args: list[str]) -> subprocess.CompletedProcess[str]:
+    try:
+        result = subprocess.run(
+            ["git", *args],
+            cwd=workdir,
+            capture_output=True,
+            text=True,
+            timeout=15,
+            check=False,
+        )
+    except FileNotFoundError as exc:
+        raise CursorMCPError(
+            "git is not installed; install git or set CURSOR_REQUIRE_GIT_REPO=false"
+        ) from exc
+    except subprocess.TimeoutExpired as exc:
+        raise CursorMCPError("git validation timed out") from exc
+
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout or "git command failed").strip()
+        raise CursorMCPError(f"git validation failed: {detail}")
+    return result

src/cursor_mcp/server.py

@@ -0,0 +1,150 @@
+"""Cursor MCP server — delegate coding work to Cursor via the Cursor SDK."""
+
+from __future__ import annotations
+
+import json
+import os
+
+from mcp.server.fastmcp import FastMCP
+
+from cursor_mcp import __version__
+from cursor_mcp.errors import CursorMCPError
+from cursor_mcp.client import delegate_once, delegate_session, get_status, resume_agent
+from cursor_mcp.config import settings
+from cursor_mcp.sessions import list_sessions
+
+mcp = FastMCP(
+    "cursor",
+    instructions=(
+        "Delegate software engineering work to Cursor's local agent via the Cursor SDK. "
+        "Use cursor_delegate for one-shot tasks (bug fix, small feature). "
+        "Use cursor_delegate_session when follow-ups are likely (multi-step refactors). "
+        "Use cursor_resume with agent_id for follow-up prompts on an existing Cursor session. "
+        "Always pass an absolute project path as cwd when known. "
+        "Requires CURSOR_API_KEY. This runs Cursor's agent loop locally against files on disk; "
+        "it does not control an open Cursor IDE window."
+    ),
+)
+
+
+def _json(data: object) -> str:
+    return json.dumps(data, indent=2, default=str)
+
+
+@mcp.tool()
+def cursor_get_config() -> str:
+    """Return cursor-mcp configuration and recent delegated sessions (no secrets)."""
+    try:
+        configured = bool(settings.api_key or os.environ.get("CURSOR_API_KEY"))
+        payload = {
+            "version": __version__,
+            "api_key_configured": configured,
+            "default_model": settings.default_model,
+            "default_cwd": settings.default_cwd,
+            "timeout_seconds": settings.timeout_seconds,
+            "require_git_repo": settings.require_git_repo,
+            "recent_sessions": list_sessions(),
+        }
+        return _json(payload)
+    except Exception as exc:
+        return _json({"status": "error", "message": str(exc)})
+
+
+@mcp.tool()
+def cursor_test_connection(cwd: str = "") -> str:
+    """Verify CURSOR_API_KEY and local project path. Does not run a full coding task."""
+    try:
+        from cursor_mcp.client import _require_api_key, _resolve_cwd
+
+        _require_api_key()
+        workdir = _resolve_cwd(cwd)
+        return _json(
+            {
+                "status": "ok",
+                "message": "CURSOR_API_KEY is set and working directory is valid.",
+                "cwd": str(workdir),
+                "default_model": settings.default_model,
+            }
+        )
+    except CursorMCPError as exc:
+        return _json({"status": "error", "message": str(exc)})
+
+
+@mcp.tool()
+def cursor_delegate(
+    prompt: str,
+    cwd: str = "",
+    model: str = "",
+    task_summary: str = "",
+) -> str:
+    """Run a one-shot coding task through Cursor's local agent.
+
+    Args:
+        prompt: Full instructions for Cursor (scope, files, acceptance criteria).
+        cwd: Absolute path to the project/repo root. Defaults to CURSOR_DEFAULT_CWD.
+        model: Cursor model id (default composer-2.5).
+        task_summary: Short label for session history.
+    """
+    try:
+        return _json(
+            delegate_once(
+                prompt=prompt,
+                cwd=cwd,
+                model=model,
+                task_summary=task_summary,
+            )
+        )
+    except CursorMCPError as exc:
+        return _json({"status": "error", "message": str(exc)})
+
+
+@mcp.tool()
+def cursor_delegate_session(
+    prompt: str,
+    cwd: str = "",
+    model: str = "",
+    task_summary: str = "",
+) -> str:
+    """Start a durable Cursor agent session for multi-step coding work.
+
+    Returns agent_id — pass it to cursor_resume for follow-up instructions.
+    Prefer this for refactors, feature builds, or tasks needing iteration.
+    """
+    try:
+        return _json(
+            delegate_session(
+                prompt=prompt,
+                cwd=cwd,
+                model=model,
+                task_summary=task_summary,
+            )
+        )
+    except CursorMCPError as exc:
+        return _json({"status": "error", "message": str(exc)})
+
+
+@mcp.tool()
+def cursor_resume(
+    agent_id: str,
+    prompt: str,
+    model: str = "",
+) -> str:
+    """Send a follow-up prompt to an existing Cursor agent session."""
+    try:
+        return _json(resume_agent(agent_id=agent_id, prompt=prompt, model=model))
+    except CursorMCPError as exc:
+        return _json({"status": "error", "message": str(exc)})
+
+
+@mcp.tool()
+def cursor_list_sessions(limit: int = 10) -> str:
+    """List recent Cursor delegation sessions with agent_id for resume."""
+    return _json({"sessions": list_sessions(limit=limit)})
+
+
+def main() -> None:
+    mcp.run()
+
+
+if __name__ == "__main__":
+    main()

src/cursor_mcp/sessions.py

@@ -0,0 +1,64 @@
+"""Persist Cursor agent IDs for follow-up delegation."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from cursor_mcp.config import settings
+
+
+def _path() -> Path:
+    path = Path(settings.sessions_path).expanduser()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def load_sessions() -> dict[str, Any]:
+    path = _path()
+    if not path.exists():
+        return {"sessions": []}
+    try:
+        return json.loads(path.read_text())
+    except json.JSONDecodeError:
+        return {"sessions": []}
+
+
+def save_session(
+    *,
+    agent_id: str,
+    run_id: str,
+    cwd: str,
+    task_summary: str,
+    status: str,
+    result_preview: str = "",
+) -> dict[str, Any]:
+    data = load_sessions()
+    entry = {
+        "agent_id": agent_id,
+        "last_run_id": run_id,
+        "cwd": cwd,
+        "task_summary": task_summary,
+        "status": status,
+        "result_preview": result_preview[:500],
+        "updated_at": datetime.now(timezone.utc).isoformat(),
+    }
+    sessions: list[dict[str, Any]] = data.setdefault("sessions", [])
+    sessions = [s for s in sessions if s.get("agent_id") != agent_id]
+    sessions.insert(0, entry)
+    data["sessions"] = sessions[:20]
+    _path().write_text(json.dumps(data, indent=2))
+    return entry
+
+
+def list_sessions(limit: int = 10) -> list[dict[str, Any]]:
+    return load_sessions().get("sessions", [])[:limit]
+
+
+def get_session(agent_id: str) -> dict[str, Any] | None:
+    for entry in load_sessions().get("sessions", []):
+        if entry.get("agent_id") == agent_id:
+            return entry
+    return None