luplo.mcp ========= .. py:module:: luplo.mcp .. autoapi-nested-parse:: 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 ---------- .. autoapisummary:: luplo.mcp.mcp Functions --------- .. autoapisummary:: luplo.mcp.luplo_work_open luplo.mcp.luplo_work_list luplo.mcp.luplo_work_resume luplo.mcp.luplo_work_close luplo.mcp.luplo_item_upsert luplo.mcp.luplo_item_search luplo.mcp.luplo_item_show luplo.mcp.luplo_check luplo.mcp.luplo_impact luplo.mcp.luplo_brief luplo.mcp.luplo_task_add luplo.mcp.luplo_task_list luplo.mcp.luplo_task_start luplo.mcp.luplo_task_done luplo.mcp.luplo_task_edit luplo.mcp.luplo_task_block luplo.mcp.luplo_capture_add luplo.mcp.luplo_capture_list luplo.mcp.luplo_capture_search luplo.mcp.luplo_capture_set_state luplo.mcp.luplo_capture_discard luplo.mcp.luplo_capture_redact luplo.mcp.luplo_capture_annotate luplo.mcp.luplo_capture_promote luplo.mcp.luplo_idea_add luplo.mcp.luplo_idea_list luplo.mcp.luplo_idea_search luplo.mcp.luplo_idea_redact luplo.mcp.luplo_qa_add luplo.mcp.luplo_qa_pass luplo.mcp.luplo_qa_fail luplo.mcp.luplo_qa_list_pending luplo.mcp.luplo_page_sync luplo.mcp.luplo_history_query luplo.mcp.luplo_save_decisions luplo.mcp.luplo_import_begin luplo.mcp.luplo_import_finalize luplo.mcp.main Module Contents --------------- .. py:data:: mcp .. py:function:: luplo_work_open(title: str, project_id: str, description: str = '', system_ids: list[str] | None = None, actor_id: str = 'claude') -> str :async: Open a new work unit to group related decisions. :param title: What this work unit is about (e.g. "Vendor system design"). :param project_id: Project to scope this work unit to. :param description: Optional longer description. :param system_ids: Systems this work touches. :param actor_id: Who is opening this (defaults to "claude"). .. py:function:: luplo_work_list(project_id: str, status: str = '') -> str :async: 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``. :param project_id: Project scope. :param status: Optional filter — "in_progress", "done", or "abandoned". Empty string returns all statuses. .. py:function:: luplo_work_resume(query: str, project_id: str) -> str :async: 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. :param query: Keyword to search in work unit titles. :param project_id: Project scope. .. py:function:: luplo_work_close(work_unit_id: str, actor_id: str = 'claude') -> str :async: Close a work unit. :param work_unit_id: ID of the work unit to close. :param actor_id: Who is closing this. .. py:function:: 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 :async: 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. :param title: Item title. :param project_id: Project scope. :param item_type: One of decision, knowledge, policy, document, research. :param body: Item body text. :param rationale: Why this decision was made. :param system_ids: Systems this item relates to. :param tags: Free-form tags. :param work_unit_id: Link to an active work unit. :param supersedes_id: ID of item this supersedes (for edits). :param source_url: Required when item_type='research' (the cached URL). Optional for other types. :param 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). :param actor_id: Who created this. .. py:function:: luplo_item_search(query: str, project_id: str, item_types: list[str] | None = None, system_ids: list[str] | None = None, limit: int = 10, tsquery: str | None = None) -> str :async: 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" :param query: Simple-dialect search string. Used unless *tsquery* is set. :param project_id: Project scope. :param item_types: Filter by item types (e.g. ["decision"]). :param system_ids: Filter by systems. :param limit: Maximum results. :param tsquery: Raw PostgreSQL ``to_tsquery`` expression. When set, bypasses the simple parser and glossary expansion entirely. .. py:function:: luplo_item_show(item_id: str, project_id: str) -> str :async: 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". :param item_id: Full UUID or ≥8-char hex prefix of the item. :param 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. .. py:function:: luplo_check(project_id: str, rule: str = '', severity: str = 'warn') -> str :async: 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. :param project_id: Project to check. :param rule: If set, run only this rule (name). Empty means all enabled rules. :param 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. .. py:function:: luplo_impact(item_id: str, project_id: str, depth: int = 5) -> str :async: 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. :param item_id: Full UUID or 8+ char hex prefix of the root item. :param project_id: Project scope. Impact is project-local. :param 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. .. py:function:: luplo_brief(project_id: str, system_id: str = '', keyword: str = '') -> str :async: Get a project context brief — active work units + recent decisions. Use this at the start of a session to load context. :param project_id: Project scope. :param system_id: Optional system filter. :param keyword: Optional keyword filter for items. .. py:function:: 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 :async: Create a new task in 'proposed' status (item_type='task'). .. py:function:: luplo_task_list(work_unit_id: str, status: str = '') -> str :async: List tasks for a work unit (chain heads, ordered by sort_order). .. py:function:: luplo_task_start(task_id: str, actor_id: str = 'claude', project_id: str = '') -> str :async: 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. .. py:function:: luplo_task_done(task_id: str, summary: str = '', actor_id: str = 'claude', project_id: str = '', propose_decision: bool = False) -> str :async: 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``. .. py:function:: luplo_task_edit(task_id: str, title: str = '', body: str = '', sort_order: int = -1, actor_id: str = 'claude', project_id: str = '') -> str :async: 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. .. py:function:: luplo_task_block(task_id: str, reason: str, actor_id: str = 'claude', project_id: str = '') -> str :async: Transition a task to 'blocked'. Auto-creates a decision item. Pass *project_id* to scope prefix resolution. .. py:function:: luplo_capture_add(text: str, actor_id: str = 'claude') -> str :async: 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. .. py:function:: luplo_capture_list(review_state: str = '', limit: int = 100, include_discarded: bool = False, include_redacted: bool = False) -> str :async: List recent captures, newest first. .. py:function:: luplo_capture_search(query: str = '', review_state: str = '', limit: int = 50, include_discarded: bool = False, include_redacted: bool = False) -> str :async: Search recent captures by text and optional state. .. py:function:: luplo_capture_set_state(capture_id: str, review_state: str, actor_id: str = 'claude') -> str :async: Move a capture to another review state. .. py:function:: luplo_capture_discard(capture_id: str, actor_id: str = 'claude') -> str :async: Discard a capture so default list/search hides it. .. py:function:: luplo_capture_redact(capture_id: str, actor_id: str = 'claude') -> str :async: Redact capture content from normal storage. .. py:function:: luplo_capture_annotate(capture_id: str, summary: str = '', sensitivity_hint: str = '', signals: dict[str, Any] | None = None) -> str :async: Store caller-supplied BYOLLM annotation hints on a capture. .. py:function:: luplo_capture_promote(capture_id: str, project_id: str, item_type: str, title: str, body: str = '', actor_id: str = 'claude') -> str :async: Explicitly promote a capture into a curated item. .. py:function:: luplo_idea_add(text: str, project_id: str, work_unit_id: str, actor_id: str = 'claude') -> str :async: 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. .. py:function:: luplo_idea_list(work_unit_id: str, project_id: str, limit: int = 100, include_redacted: bool = False) -> str :async: 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. .. py:function:: luplo_idea_search(project_id: str, query: str = '', tsquery: str = '', work_unit_id: str = '', author: str = '', since: str = '', until: str = '', include_redacted: bool = False, limit: int = 50) -> str :async: 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="", since="this_quarter", ) .. py:function:: luplo_idea_redact(idea_id: str, project_id: str, actor_id: str = 'claude') -> str :async: 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. .. py:function:: 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 :async: Create a qa_check in 'pending' status. coverage = auto_partial|human_only. .. py:function:: luplo_qa_pass(qa_id: str, evidence: str = '', actor_id: str = 'claude', project_id: str = '') -> str :async: Transition a qa_check to 'passed' with optional evidence. Pass *project_id* to scope prefix resolution. .. py:function:: luplo_qa_fail(qa_id: str, reason: str, actor_id: str = 'claude', project_id: str = '') -> str :async: Transition a qa_check to 'failed'. Pass *project_id* to scope prefix resolution. .. py:function:: luplo_qa_list_pending(project_id: str, task_id: str = '', item_id: str = '', work_unit_id: str = '') -> str :async: List pending qa_checks. Filter by task / item / work_unit (one of). .. py:function:: luplo_page_sync(source_type: str, source_page_id: str, full_content: str, project_id: str, source_event_id: str = '') -> str :async: 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. :param source_type: Origin system (e.g. "notion", "confluence"). :param source_page_id: External page identifier. :param full_content: Full page content (markdown). :param project_id: Project scope. :param source_event_id: Optional external event ID for idempotency. .. py:function:: luplo_history_query(project_id: str, item_id: str = '', since: str = '', semantic_impacts: list[str] | None = None, limit: int = 20) -> str :async: Query change history for items. :param project_id: Project scope. :param item_id: Optional specific item to query. :param since: Optional ISO timestamp to filter from. :param semantic_impacts: Filter by impact types (e.g. ["numeric_change", "rule_addition"]). :param limit: Maximum entries. .. py:function:: luplo_save_decisions(transcript: str, project_id: str, work_unit_id: str = '', actor_id: str = 'claude') -> str :async: Extract and save decisions from a conversation transcript. v0.5: Returns a stub response. LLM extraction integration is planned for post-v0.5. :param transcript: Conversation text to extract decisions from. :param project_id: Project scope. :param work_unit_id: Optional work unit to attach extracted items to. :param actor_id: Who authored the transcript. .. py:function:: 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] :async: 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. :param project_id: Project owning the bundle. :param 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. :param dest_lang: ISO 639-1 target language. ``None`` preserves source language verbatim. :param force: When True, archive any prior import for the same content set and stage a fresh bundle. Default False — duplicates return a refusal payload. :param 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). :param 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``. .. py:function:: 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] :async: 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. :param bundle_id: Work-unit id returned by ``luplo_import_begin``. :param items: List of result-item dicts to persist as luplo items. :param project_id: Project owning the bundle. A cross-project guard rejects mismatches. :param close_work_unit: When True, transition the bundle's work unit to ``done`` after items are written. :param 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``. .. py:function:: main() -> None Run the MCP server over stdio.