ten31-database/backend/mcp/server.py

#!/usr/bin/env python3
"""Ten31 CRM MCP server (Workstream C).

Exposes CRM reads, retrieval modes, and logged writes to the Claude Agent SDK
over MCP (stdio). All logic lives in crm_tools.py (tested independently); this
file is the thin transport wrapper.

Run:
    pip install mcp            # one-time (MCP Python SDK)
    CRM_DB_PATH=/data/crm.db python3 backend/mcp/server.py

Register with the Agent SDK / Claude Code as an stdio MCP server pointing at this
script. NO outbound/contact tools are exposed — that capability is gated to
Phase 3 behind the compliance review (CLAUDE.md guardrails #4, #6).
"""
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import crm_tools as t  # noqa: E402
import architect_tools as at  # noqa: E402

from mcp.server.fastmcp import FastMCP  # noqa: E402

mcp = FastMCP("ten31-crm")


# ── reads ──
@mcp.tool()
def get_entity(lp_id: str) -> dict:
    """Fetch a canonical LP/organization/person entity by id, with its linked
    source records and interaction count."""
    return t.get_entity(lp_id)


@mcp.tool()
def search_records(query: str = "", entity_kind: str = "", limit: int = 20) -> dict:
    """Structured search over canonical entities by name substring and kind
    ('lp' | 'organization' | 'person')."""
    return t.search_records(query=query or None, entity_kind=entity_kind or None, limit=limit)


@mcp.tool()
def get_interaction_history(lp_id: str, limit: int = 20) -> dict:
    """Merged, dated interaction history (communications + fundraising grid notes)
    for a canonical entity."""
    return t.get_interaction_history(lp_id, limit=limit)


@mcp.tool()
def get_investor_contacts(lp_id: str) -> dict:
    """List all contacts (people) belonging to an investor entity — the
    one-investor-to-many-contacts relationship (e.g. a family office's several people)."""
    return t.get_investor_contacts(lp_id)


# ── retrieval modes ──
@mcp.tool()
def hybrid_search(query: str, top_k: int = 8, lp_id: str = "", doc_type: str = "",
                  date_from: int = 0, date_to: int = 0) -> dict:
    """Dense + BM25 + rerank retrieval (default; best for entity-heavy queries).
    Optional filters: lp_id, doc_type, date_from/date_to (epoch seconds)."""
    return t.hybrid_search(query, top_k=top_k, lp_id=lp_id or None, doc_type=doc_type or None,
                           date_from=date_from or None, date_to=date_to or None)


@mcp.tool()
def semantic_search(query: str, top_k: int = 8, lp_id: str = "", doc_type: str = "") -> dict:
    """Dense-only retrieval (high recall)."""
    return t.semantic_search(query, top_k=top_k, lp_id=lp_id or None, doc_type=doc_type or None)


@mcp.tool()
def keyword_search(query: str, top_k: int = 8, lp_id: str = "", doc_type: str = "") -> dict:
    """High-precision lexical retrieval (sparse leg + rerank)."""
    return t.keyword_search(query, top_k=top_k, lp_id=lp_id or None, doc_type=doc_type or None)


# ── writes (logged) ──
@mcp.tool()
def log_interaction(action: str, actor_type: str = "agent", actor_id: str = "",
                    target_id: str = "", payload: dict = None, source: str = "mcp") -> dict:
    """Append an entry to the append-only interaction log (guardrail #5)."""
    return t.log_interaction(action, actor_type=actor_type, actor_id=actor_id or None,
                             target_id=target_id or None, payload=payload, source=source)


@mcp.tool()
def set_entity_enrichment(lp_id: str, fields: dict, actor_id: str = "analyst") -> dict:
    """One-way enrichment write into a canonical entity (thesis_fit, segment,
    warmth_score, accreditation_status, etc.). Logged automatically."""
    return t.set_entity_enrichment(lp_id, fields, actor_id=actor_id)


# ── Architect thesis tools (Phase 1; drafts only — no approve/promote here) ──
@mcp.tool()
def get_thesis(line_key: str) -> dict:
    """Fetch a thesis line and its node tree (throughline → sections → claims → proof-points)."""
    return at.get_thesis(line_key)


@mcp.tool()
def list_thesis_lines() -> dict:
    """List all thesis lines (the core spine + per-segment lines)."""
    return at.list_thesis_lines()


@mcp.tool()
def get_canonical_thesis(line_key: str) -> dict:
    """The current partner-APPROVED canonical thesis for a line. Fails closed if none approved."""
    return at.get_canonical_thesis(line_key)


@mcp.tool()
def get_review_feedback(version_id: str) -> dict:
    """Partners' reviews/feedback on a thesis version — what to iterate on."""
    return at.get_review_feedback(version_id)


@mcp.tool()
def create_thesis_line(line_key: str, name: str, segment_key: str = "", is_core: bool = False,
                       description: str = "") -> dict:
    """Create a new thesis line (a narrative, e.g. the core spine or a per-segment line)."""
    return at.create_thesis_line(line_key, name, segment_key=segment_key or None,
                                 is_core=is_core, description=description or None)


@mcp.tool()
def upsert_thesis_node(line_id: str, node_type: str, body: str, title: str = "", parent_id: str = "",
                       node_id: str = "", variant_group: str = "", change_reason: str = "") -> dict:
    """Create or edit a thesis node (a claim, section, proof-point, etc.). Edits are revisioned."""
    return at.upsert_thesis_node(line_id, node_type, body, title=title or None,
                                 parent_id=parent_id or None, node_id=node_id or None,
                                 variant_group=variant_group or None, change_reason=change_reason or None)


@mcp.tool()
def create_thesis_version(line_key: str, rationale: str = "") -> dict:
    """Freeze the current node tree into an immutable DRAFT version (stays draft until a human approves)."""
    return at.create_thesis_version(line_key, rationale=rationale or None)


@mcp.tool()
def submit_version_for_review(version_id: str) -> dict:
    """Move a draft thesis version to 'in_review' so the partners can weigh in. Cannot make it canonical."""
    return at.submit_version_for_review(version_id)


@mcp.tool()
def list_segments() -> dict:
    """List active LP segment definitions."""
    return at.list_segments()


@mcp.tool()
def upsert_segment(segment_key: str, name: str, definition: str = "", needs_to_hear: str = "",
                   avoid: str = "") -> dict:
    """Create/replace an LP segment's active definition."""
    return at.upsert_segment(segment_key, name, definition=definition or None,
                             needs_to_hear=needs_to_hear or None, avoid=avoid or None)


if __name__ == "__main__":
    mcp.run()