luplo.mcp

luplo MCP server — stdio interface for Claude Desktop/Code.

Run with:

uv run python -m luplo.mcp

Or configure in Claude Desktop’s claude_desktop_config.json:

{
  "mcpServers": {
    "luplo": {
      "command": "uv",
      "args": ["run", "python", "-m", "luplo.mcp"],
      "env": {"LUPLO_DB_URL": "postgresql://localhost/luplo"}
    }
  }
}

Attributes

mcp

Functions

luplo_work_open(→ str)

Open a new work unit to group related decisions.

luplo_work_list(→ str)

List work units in a project, ordered by created_at DESC.

luplo_work_resume(→ str)

Find in-progress work units by title keyword + their open tasks/QA.

luplo_work_close(→ str)

Close a work unit.

luplo_item_upsert(→ str)

Create or update (supersede) an item.

luplo_item_search(→ str)

Search items using glossary-expanded full-text search.

luplo_item_show(→ str)

Return the full body, rationale, and metadata of a single item.

luplo_check(→ str)

Run the deterministic rule pack and return findings as markdown.

luplo_impact(→ str)

Traverse typed edges to find an item's blast radius.

luplo_brief(→ str)

Get a project context brief — active work units + recent decisions.

luplo_task_add(→ str)

Create a new task in 'proposed' status (item_type='task').

luplo_task_list(→ str)

List tasks for a work unit (chain heads, ordered by sort_order).

luplo_task_start(→ str)

Transition a task to 'in_progress'. Fails if another task is in_progress.

luplo_task_done(→ str)

Transition a task to 'done'.

luplo_task_edit(→ str)

Edit a task's title / body / sort_order via supersede.

luplo_task_block(→ str)

Transition a task to 'blocked'. Auto-creates a decision item.

luplo_capture_add(→ str)

Save raw text into the capture backlog.

luplo_capture_list(→ str)

List recent captures, newest first.

luplo_capture_search(→ str)

Search recent captures by text and optional state.

luplo_capture_set_state(→ str)

Move a capture to another review state.

luplo_capture_discard(→ str)

Discard a capture so default list/search hides it.

luplo_capture_redact(→ str)

Redact capture content from normal storage.

luplo_capture_annotate(→ str)

Store caller-supplied BYOLLM annotation hints on a capture.

luplo_capture_promote(→ str)

Explicitly promote a capture into a curated item.

luplo_idea_add(→ str)

Append an ideation note to a work unit (append-only, redact-only).

luplo_idea_list(→ str)

List ideas for a work unit (newest first).

luplo_idea_search(→ str)

Full-text search over ideas in a project (optionally narrowed by WU).

luplo_idea_redact(→ str)

Mark an idea redacted (idempotent — no-op if already redacted).

luplo_qa_add(→ str)

Create a qa_check in 'pending' status. coverage = auto_partial|human_only.

luplo_qa_pass(→ str)

Transition a qa_check to 'passed' with optional evidence.

luplo_qa_fail(→ str)

Transition a qa_check to 'failed'.

luplo_qa_list_pending(→ str)

List pending qa_checks. Filter by task / item / work_unit (one of).

luplo_page_sync(→ str)

Queue an external page for sync (debounced).

luplo_history_query(→ str)

Query change history for items.

luplo_save_decisions(→ str)

Extract and save decisions from a conversation transcript.

luplo_import_begin(→ dict[str, Any])

Stage a new import bundle from spec/plan markdown content.

luplo_import_finalize(→ dict[str, Any])

Commit agent-extracted items into the staged import bundle.

main(→ None)

Run the MCP server over stdio.

Module Contents

luplo.mcp.mcp
async luplo.mcp.luplo_work_open(title: str, project_id: str, description: str = '', system_ids: list[str] | None = None, actor_id: str = 'claude') → str

Open a new work unit to group related decisions.

Parameters:

  • title – What this work unit is about (e.g. “Vendor system design”).

  • project_id – Project to scope this work unit to.

  • description – Optional longer description.

  • system_ids – Systems this work touches.

  • actor_id – Who is opening this (defaults to “claude”).

async luplo.mcp.luplo_work_list(project_id: str, status: str = '') → str

List work units in a project, ordered by created_at DESC.

Use this to recall recently closed work units (e.g. to re-open a follow-up task) or to inspect the backlog of in_progress units. For keyword search across in_progress units, prefer luplo_work_resume.

Parameters:

  • project_id – Project scope.

  • status – Optional filter — “in_progress”, “done”, or “abandoned”. Empty string returns all statuses.

async luplo.mcp.luplo_work_resume(query: str, project_id: str) → str

Find in-progress work units by title keyword + their open tasks/QA.

Returns a markdown brief per matching work unit including the current in_progress task, pending tasks (sort_order ASC), and pending qa_checks.

Parameters:

  • query – Keyword to search in work unit titles.

  • project_id – Project scope.

async luplo.mcp.luplo_work_close(work_unit_id: str, actor_id: str = 'claude') → str

Close a work unit.

Parameters:

  • work_unit_id – ID of the work unit to close.

  • actor_id – Who is closing this.

async luplo.mcp.luplo_item_upsert(title: str, project_id: str, item_type: str = 'decision', body: str = '', rationale: str = '', system_ids: list[str] | None = None, tags: list[str] | None = None, work_unit_id: str = '', supersedes_id: str = '', source_url: str = '', expires_at: str = '', actor_id: str = 'claude') → str

Create or update (supersede) an item.

To edit an existing item, pass its ID as supersedes_id — a new version is created and the old one is preserved.

Parameters:

  • title – Item title.

  • project_id – Project scope.

  • item_type – One of decision, knowledge, policy, document, research.

  • body – Item body text.

  • rationale – Why this decision was made.

  • system_ids – Systems this item relates to.

  • tags – Free-form tags.

  • work_unit_id – Link to an active work unit.

  • supersedes_id – ID of item this supersedes (for edits).

  • source_url – Required when item_type=’research’ (the cached URL). Optional for other types.

  • expires_at – ISO-8601 timestamp for cache expiry. When omitted and item_type=’research’, defaults to now + research_ttl_days from config (90 days default).

  • actor_id – Who created this.

Search items using glossary-expanded full-text search.

Two modes:

  1. Simple (query=) — small dialect designed for humans: plain words AND together, "exact phrase", OR (uppercase), -negation. Glossary auto-expands required terms (e.g. “vendor” also matches “shop”, “NPC merchant”). No parentheses, no prefix, no &|! operators — those are passed through as literal chars.

  2. Raw tsquery (tsquery=) — escape hatch for callers (LLMs, advanced users) that want PostgreSQL’s full to_tsquery grammar: & (AND), | (OR), ! (NOT), :* (prefix), <-> (phrase distance), parentheses for grouping. Glossary is not applied — caller is responsible for synonym coverage. Bad syntax raises an error.

Pick one — when tsquery is set, query is ignored.

Examples:

# Simple
query="inventory slot OR 인벤토리"
query="\"InventoryWindow\" -deprecated"

# Raw tsquery
tsquery="(inventory | 인벤토리) & slot & !deprecated"
tsquery="hearth:* & ward & decision"
Parameters:

  • query – Simple-dialect search string. Used unless tsquery is set.

  • project_id – Project scope.

  • item_types – Filter by item types (e.g. [“decision”]).

  • system_ids – Filter by systems.

  • limit – Maximum results.

  • tsquery – Raw PostgreSQL to_tsquery expression. When set, bypasses the simple parser and glossary expansion entirely.

async luplo.mcp.luplo_item_show(item_id: str, project_id: str) → str

Return the full body, rationale, and metadata of a single item.

Use this when a luplo_item_search preview is truncated and the full text is needed to reason about a decision, knowledge entry, or document. Accepts a full UUID or ≥8-char hex prefix; prefix lookups are scoped to project_id. Soft-deleted items return “not found”.

Parameters:

  • item_id – Full UUID or ≥8-char hex prefix of the item.

  • project_id – Project scope for prefix disambiguation and safety.

Returns:

Markdown text with a title header, a metadata block, and untruncated Body / Rationale sections. Empty sections are skipped.

async luplo.mcp.luplo_check(project_id: str, rule: str = '', severity: str = 'warn') → str

Run the deterministic rule pack and return findings as markdown.

Rules are project-local, SQL+Python only, and never call an LLM. The set is fixed per release — there is no plugin runtime.

Parameters:

  • project_id – Project to check.

  • rule – If set, run only this rule (name). Empty means all enabled rules.

  • severity – Show findings at or above this level — “error”, “warn” (default), or “info”.

Returns:

Markdown listing each finding with its severity, rule name, item id prefix, and message. Empty state returns a short “no findings” line.

async luplo.mcp.luplo_impact(item_id: str, project_id: str, depth: int = 5) → str

Traverse typed edges to find an item’s blast radius.

Walks outgoing depends / blocks / supersedes / conflicts edges from the given item up to depth hops and returns every reachable item, each labelled with the edge type that first reached it at its shortest-path depth. Traversal never crosses project boundaries; cycles are broken automatically; items appear once.

Depth is capped at 5 server-side. Requesting deeper is not supported and is not a performance limit — it reflects a model-hygiene principle. If impact analysis needs more than five hops, the model should be decomposed.

Parameters:

  • item_id – Full UUID or 8+ char hex prefix of the root item.

  • project_id – Project scope. Impact is project-local.

  • depth – Max traversal depth, 1..5. Default 5.

Returns:

Markdown text listing the root and each impacted item with its depth and edge type. Item IDs are 8-char prefixes suitable for citing back in follow-up tool calls.

async luplo.mcp.luplo_brief(project_id: str, system_id: str = '', keyword: str = '') → str

Get a project context brief — active work units + recent decisions.

Use this at the start of a session to load context.

Parameters:

  • project_id – Project scope.

  • system_id – Optional system filter.

  • keyword – Optional keyword filter for items.

async luplo.mcp.luplo_task_add(title: str, project_id: str, work_unit_id: str, body: str = '', system_ids: list[str] | None = None, sort_order: int | None = None, actor_id: str = 'claude') → str

Create a new task in ‘proposed’ status (item_type=’task’).

async luplo.mcp.luplo_task_list(work_unit_id: str, status: str = '') → str

List tasks for a work unit (chain heads, ordered by sort_order).

async luplo.mcp.luplo_task_start(task_id: str, actor_id: str = 'claude', project_id: str = '') → str

Transition a task to ‘in_progress’. Fails if another task is in_progress.

Pass project_id (the active project) whenever it is known. It scopes prefix resolution so a short task_id prefix cannot accidentally match a task in a different project.

async luplo.mcp.luplo_task_done(task_id: str, summary: str = '', actor_id: str = 'claude', project_id: str = '', propose_decision: bool = False) → str

Transition a task to ‘done’.

Pass project_id to scope prefix resolution. When propose_decision is true, the response is followed by a draft decision item derived from this task — never inserted. The draft is a suggestion for the caller to show the human, who then decides whether to save it via luplo_item_upsert.

async luplo.mcp.luplo_task_edit(task_id: str, title: str = '', body: str = '', sort_order: int = -1, actor_id: str = 'claude', project_id: str = '') → str

Edit a task’s title / body / sort_order via supersede.

Status is preserved (call luplo_task_start / luplo_task_done etc. to change status). Pass empty strings / sort_order=-1 for fields you do not want to change. project_id scopes prefix resolution.

async luplo.mcp.luplo_task_block(task_id: str, reason: str, actor_id: str = 'claude', project_id: str = '') → str

Transition a task to ‘blocked’. Auto-creates a decision item.

Pass project_id to scope prefix resolution.

async luplo.mcp.luplo_capture_add(text: str, actor_id: str = 'claude') → str

Save raw text into the capture backlog.

This tool is BYOLLM. The caller may summarize or classify before calling, but luplo core only stores the provided text.

async luplo.mcp.luplo_capture_list(review_state: str = '', limit: int = 100, include_discarded: bool = False, include_redacted: bool = False) → str

List recent captures, newest first.

Search recent captures by text and optional state.

async luplo.mcp.luplo_capture_set_state(capture_id: str, review_state: str, actor_id: str = 'claude') → str

Move a capture to another review state.

async luplo.mcp.luplo_capture_discard(capture_id: str, actor_id: str = 'claude') → str

Discard a capture so default list/search hides it.

async luplo.mcp.luplo_capture_redact(capture_id: str, actor_id: str = 'claude') → str

Redact capture content from normal storage.

async luplo.mcp.luplo_capture_annotate(capture_id: str, summary: str = '', sensitivity_hint: str = '', signals: dict[str, Any] | None = None) → str

Store caller-supplied BYOLLM annotation hints on a capture.

async luplo.mcp.luplo_capture_promote(capture_id: str, project_id: str, item_type: str, title: str, body: str = '', actor_id: str = 'claude') → str

Explicitly promote a capture into a curated item.

async luplo.mcp.luplo_idea_add(text: str, project_id: str, work_unit_id: str, actor_id: str = 'claude') → str

Append an ideation note to a work unit (append-only, redact-only).

Deprecated for raw intake. Use capture for unstructured backlog entries. Ideas remain for compatibility with work-unit-scoped ideation notes.

Use for half-formed thoughts, exploration trails, “what if” notes — anything you want to remember but is not yet a committed decision. Mistakes are recovered via luplo_idea_redact; ideas are never deleted. Closed (done) work units accept ideas — useful for retro notes. Archived / abandoned work units reject new ideas.

async luplo.mcp.luplo_idea_list(work_unit_id: str, project_id: str, limit: int = 100, include_redacted: bool = False) → str

List ideas for a work unit (newest first).

Deprecated for raw intake. Use capture for unstructured backlog entries. Ideas remain for compatibility with work-unit-scoped ideation notes.

project_id is required to prevent an 8-char prefix collision with another project’s WU from matching, and to add a defence-in-depth predicate at the SQL level. Without it, a full UUID from another project would surface that project’s ideas. Mutually consistent with the other luplo_idea_* tools.

Redacted ideas are excluded by default. include_redacted=True surfaces their existence (id + timestamp + [REDACTED] marker) for audit flows, but the original text is masked as [redacted]. Combining include_redacted=True with a text query on luplo_idea_search is rejected — fetching raw redacted content requires a SaaS-side admin path.

Full-text search over ideas in a project (optionally narrowed by WU).

Deprecated for raw intake. Use capture for unstructured backlog entries. Ideas remain for compatibility with work-unit-scoped ideation notes.

query and tsquery are mutually exclusive — pass at most one. Glossary expansion applies to query. Use tsquery for the raw to_tsquery escape hatch (no glossary).

Time filters: since accepts ISO datetime, Nd / Nw (last N days / weeks), or anchor (this_week / this_month / this_quarter). until accepts ISO datetime or Nd / Nw; anchors are since-only (returning the start of the period as until would filter out the entire current period).

author takes a full actor id. (Caller resolves “me” itself.)

Redacted ideas are excluded by default. include_redacted=True surfaces redacted rows for audit but cannot be combined with a text query — match/no-match against redacted text would leak the body via a keyword oracle. Use filter-only mode (author/since/wu) for audit flows; raw redacted text requires a SaaS-side admin path.

Caller LLMs should decompose natural-language queries into the appropriate query / since / author parameters on the client side. This tool itself is deterministic.

Worked examples:

  1. User says: “지난주 OAuth 리프레시 토큰 관련 아이디어” → call:

    luplo_idea_search(
    project_id=...,
    query="OAuth refresh token 리프레시 토큰",
    since="7d",
)

  2. User says: “내가 이번 분기에 적은 검색 인프라 idea” → resolve “내가” to the caller’s actor_id, then call:

    luplo_idea_search(
    project_id=...,
    query="검색 인프라",
    author="<my-actor-id>",
    since="this_quarter",
)
async luplo.mcp.luplo_idea_redact(idea_id: str, project_id: str, actor_id: str = 'claude') → str

Mark an idea redacted (idempotent — no-op if already redacted).

Deprecated for raw intake. Use capture for unstructured backlog entries. Ideas remain for compatibility with work-unit-scoped ideation notes.

Hides the idea from default list/search results. The row and text column are preserved (audit metadata, not deletion); redacted_at and redacted_by are stamped. The default luplo_idea_list / luplo_idea_search will not return this row. include_redacted=True on those tools surfaces the row’s existence (id + timestamp + [REDACTED] marker) but the original text is masked as [redacted]. Combining include_redacted=True with a text query is rejected to close a keyword-oracle. Raw redacted content requires a SaaS-side admin path.

Use this for mistakes or content that shouldn’t surface in normal flows; do not use this as a secret-scrubbing mechanism — the text remains in the database.

project_id is required and is enforced both in prefix resolution and at the SQL UPDATE predicate so a full UUID from another project cannot mutate this row.

async luplo.mcp.luplo_qa_add(title: str, project_id: str, coverage: str, areas: list[str] | None = None, target_task_ids: list[str] | None = None, target_item_ids: list[str] | None = None, work_unit_id: str = '', actor_id: str = 'claude') → str

Create a qa_check in ‘pending’ status. coverage = auto_partial|human_only.

async luplo.mcp.luplo_qa_pass(qa_id: str, evidence: str = '', actor_id: str = 'claude', project_id: str = '') → str

Transition a qa_check to ‘passed’ with optional evidence.

Pass project_id to scope prefix resolution.

async luplo.mcp.luplo_qa_fail(qa_id: str, reason: str, actor_id: str = 'claude', project_id: str = '') → str

Transition a qa_check to ‘failed’.

Pass project_id to scope prefix resolution.

async luplo.mcp.luplo_qa_list_pending(project_id: str, task_id: str = '', item_id: str = '', work_unit_id: str = '') → str

List pending qa_checks. Filter by task / item / work_unit (one of).

async luplo.mcp.luplo_page_sync(source_type: str, source_page_id: str, full_content: str, project_id: str, source_event_id: str = '') → str

Queue an external page for sync (debounced).

Used by Notion/Confluence custom agents to push page updates. The sync job is debounced — rapid consecutive calls for the same page are merged into one.

Parameters:

  • source_type – Origin system (e.g. “notion”, “confluence”).

  • source_page_id – External page identifier.

  • full_content – Full page content (markdown).

  • project_id – Project scope.

  • source_event_id – Optional external event ID for idempotency.

async luplo.mcp.luplo_history_query(project_id: str, item_id: str = '', since: str = '', semantic_impacts: list[str] | None = None, limit: int = 20) → str

Query change history for items.

Parameters:

  • project_id – Project scope.

  • item_id – Optional specific item to query.

  • since – Optional ISO timestamp to filter from.

  • semantic_impacts – Filter by impact types (e.g. [“numeric_change”, “rule_addition”]).

  • limit – Maximum entries.

async luplo.mcp.luplo_save_decisions(transcript: str, project_id: str, work_unit_id: str = '', actor_id: str = 'claude') → str

Extract and save decisions from a conversation transcript.

v0.5: Returns a stub response. LLM extraction integration is planned for post-v0.5.

Parameters:

  • transcript – Conversation text to extract decisions from.

  • project_id – Project scope.

  • work_unit_id – Optional work unit to attach extracted items to.

  • actor_id – Who authored the transcript.

async luplo.mcp.luplo_import_begin(project_id: str, sources: list[dict[str, str]], dest_lang: str | None = None, force: bool = False, repo_root: str = '', actor_id: str = 'claude') → dict[str, Any]

Stage a new import bundle from spec/plan markdown content.

The agent reads the spec/plan markdown files before invoking this tool and passes their contents inline via sources. This keeps the pipeline filesystem-free on the server, which is what allows it to run on the multi-tenant cloud MCP. Local-mode invocations follow the same pattern for consistency.

Returns one of two response shapes:

  • manifest (success): the full ImportManifest JSON. The agent must now read the manifest, extract candidate items per the protocol rules, verify each candidate’s status against the repository code (using parallel small-model subagents recommended), then call luplo_import_finalize with the assembled results.

  • refusal (status="refused"): a 3-layer self-documenting refusal carrying why, override, and agent_hint fields. The agent MUST surface the refusal to the user and explicitly ask before retrying with force=true. Do not silently retry.

Rules the agent must enforce on extracted items:

  • Chunk meaningfully (decisions/knowledge granularity, not per-checkbox).

  • No fenced code blocks in body — replace with a placeholder.

  • Translate to dest_lang if set; preserve source language otherwise.

  • Rejected proposals with known rationale → standalone decision items.

Parameters:

  • project_id – Project owning the bundle.

  • sources –

    Non-empty list of source files. Each entry is a dict with three string keys:

    • kind: "spec" or "plan".

    • path: caller-supplied identifier (filesystem path or other label). Stored verbatim; never opened by the server.

    • content: the file’s UTF-8 markdown content.

    At most one "spec" and one "plan" may be supplied.

  • dest_lang – ISO 639-1 target language. None preserves source language verbatim.

  • force – When True, archive any prior import for the same content set and stage a fresh bundle. Default False — duplicates return a refusal payload.

  • repo_root – Repository root path the agent will use during code verification. Stored verbatim in the manifest; the server never resolves or opens it. Empty string is allowed when verification will run elsewhere (e.g. the cloud).

  • actor_id – UUID of the actor opening the bundle. The literal "claude" resolves to the configured .luplo actor in local mode; in remote/cloud mode the server resolves the actor from the bearer token regardless.

Returns:

Either a manifest dict (with bundle_id, dest_lang, sources, protocol keys) or a refusal dict with status="refused" plus why / override / agent_hint.

async luplo.mcp.luplo_import_finalize(bundle_id: str, items: list[dict[str, Any]], project_id: str, close_work_unit: bool = False, actor_id: str = 'claude') → dict[str, Any]

Commit agent-extracted items into the staged import bundle.

items must be a list of objects matching the ResultItem shape: {item_type, title, body, status, evidence_paths, [rationale], [tags], [system_ids], [source_url]}.

Validation aborts the entire commit on protocol violations (no fenced code blocks; status=done|partial requires evidence_paths; item_type ∈ {decision, knowledge, document}). Defense-in-depth strips fenced code blocks even when the agent followed the rule and surfaces them as warnings.

Parameters:

  • bundle_id – Work-unit id returned by luplo_import_begin.

  • items – List of result-item dicts to persist as luplo items.

  • project_id – Project owning the bundle. A cross-project guard rejects mismatches.

  • close_work_unit – When True, transition the bundle’s work unit to done after items are written.

  • actor_id – UUID of the actor finalising. The literal "claude" resolves to the configured .luplo actor.

Returns:

Summary dict with status, bundle_id, items_created, and a (possibly empty) list of warnings.

luplo.mcp.main() → None

Run the MCP server over stdio.