mcp_server.py

"""Session RAG — MCP Server wrapper.

Thin MCP layer over the session RAG backend (Docker container on port 8083).
All tool names prefixed with session_ to avoid collision with obsidian-rag.
"""

import os
from pathlib import Path
import httpx
from mcp.server.fastmcp import FastMCP

RAG_URL = os.getenv("RAG_URL", "http://localhost:8083")

mcp = FastMCP(
    "Session RAG",
    instructions=(
        "RAG server for Claude Code session transcripts. Indexes distilled session "
        "markdown notes and provides hybrid search (vector + BM25 + reranking). "
        "Use session_search_hybrid for general questions about past sessions, "
        "session_search_keyword for exact terms (session IDs, dates, tool names), "
        "and session_get_full_note for complete session transcripts. "
        "Use session_search_files for a complete list of sessions matching a query. "
        "Use session_search_aggregate to see how sessions distribute across projects, tags, or time."
    ),
)


def _call(method: str, path: str, json_body: dict | None = None) -> dict:
    """Call the session RAG API backend."""
    try:
        with httpx.Client(timeout=30) as client:
            if method == "GET":
                r = client.get(f"{RAG_URL}{path}")
            else:
                r = client.post(f"{RAG_URL}{path}", json=json_body or {})
            r.raise_for_status()
            return r.json()
    except httpx.ConnectError:
        return {"error": f"Session RAG container is not running. Start with: cd {Path(__file__).parent} && docker compose up -d"}
    except Exception as e:
        return {"error": str(e)}


@mcp.tool()
def session_search_hybrid(query: str, n_results: int = 5) -> dict:
    """Hybrid search across session transcripts: vector + BM25 + reranking.

    Best for complex questions about past sessions, decisions, or work history.

    Args:
        query: Search query in natural language
        n_results: Number of results (default 5)
    """
    return _call("POST", "/search_hybrid", {"query": query, "n_results": n_results})


@mcp.tool()
def session_search_semantic(query: str, n_results: int = 5) -> dict:
    """Semantic search across session transcripts.

    Best for conceptual questions about past work.

    Args:
        query: Natural language question
        n_results: Number of results (default 5)
    """
    return _call("POST", "/search_semantic", {"query": query, "n_results": n_results})


@mcp.tool()
def session_search_keyword(query: str, n_results: int = 5) -> dict:
    """BM25 keyword search across session transcripts.

    Best for session IDs, dates, tool names, exact terms.

    Args:
        query: Exact terms to search for
        n_results: Number of results (default 5)
    """
    return _call("POST", "/search_keyword", {"query": query, "n_results": n_results})


@mcp.tool()
def session_search_files(query: str) -> dict:
    """Exhaustive file search: returns ALL session notes matching the query.

    Args:
        query: Keywords to search for
    """
    return _call("POST", "/search_files", {"query": query})


@mcp.tool()
def session_search_aggregate(query: str, group_by: str = "folder") -> dict:
    """Aggregate session search results by metadata field.

    Args:
        query: Keywords to search for
        group_by: One of: folder, tags, type, status, created_month, project
    """
    return _call("POST", "/search_aggregate", {"query": query, "group_by": group_by})


@mcp.tool()
def session_get_full_note(note_path: str) -> dict:
    """Retrieve the full content of a session transcript note.

    Args:
        note_path: Relative path (e.g., "obsidian/2026-04-03_62d83694.md")
    """
    return _call("POST", "/get_full_note", {"note_path": note_path})


@mcp.tool()
def session_list_topics() -> dict:
    """List all projects in the sessions vault with note counts."""
    return _call("GET", "/list_topics")


@mcp.tool()
def session_reindex() -> dict:
    """Re-index the sessions vault. Run after converting new sessions."""
    return _call("POST", "/index")


@mcp.tool()
def session_rag_status() -> dict:
    """Get session RAG system status: model, chunk count, config."""
    return _call("GET", "/status")


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