ChristianMCP / server.py
sreenathvemula's picture
Claude Opus 5.5
library_search: drop results below a reranker relevance floor
f5e2bd2 verified
Raw History Blame Contribute Delete
16.9 kB
"""christian-corpus MCP server β€” classic public-domain Bible study reference.
Seventeen tools over a local reference library: 23 classic commentaries, 4
public-domain translations (KJV/ASV/WEB/BSB), per-word Hebrew/Greek
morphology, 6 lexicons (Strong's, Thayer's, LSJ, BDB, AHLB, LXX equivalents),
Treasury of Scripture Knowledge cross-references, Nave's/Torrey's topics,
ISBE + McClintock encyclopedias, historic creeds/confessions, and the UBS
parallel-passage database. All data loads lazily and caches in-process.
Transports:
python server.py # Streamable HTTP on 0.0.0.0:$PORT (default 8000)
python server.py --stdio # stdio, for a local MCP client
Environment variables (HTTP mode):
PORT listen port (default 8000)
HOST bind address (default 0.0.0.0)
MCP_SECRET optional secret path segment. If set, the endpoint becomes
https://<host>/<MCP_SECRET>/mcp instead of /mcp β€” a simple
way to keep a personal deployment private without OAuth.
"""
import functools
import json
import os
import sys
import time
from typing import Optional
from mcp.server.fastmcp import FastMCP
import lookup_commentary as _commentary
import lookup_passage as _passage
import lookup_lexicon as _lexicon
import lookup_topic as _topic
import lookup_encyclopedia as _encyclopedia
import tsk_lookup as _tsk
import lookup_search as _search
import lookup_word_occurrences as _word_occ
import lookup_historical_documents as _historical
import lookup_parallel as _parallel
import lookup_interlinear as _interlinear
import lookup_versification as _versification
import lookup_original_search as _original_search
from query_log import log_call
_TRANSPORT_MODE = "http" # flipped to "stdio" in __main__ when run that way
_PRIVATE_MODE = "--private" in sys.argv
def _logged(name):
"""Records every call (args, ok/error, size, timing) to data/usage_log.db
-- what real callers actually ask, for corpus-priority analysis. See
query_log.py. A logging failure never breaks the underlying tool call."""
def deco(fn):
@functools.wraps(fn)
def wrapper(*args, **kwargs):
t0 = time.time()
try:
result = fn(*args, **kwargs)
log_call(_TRANSPORT_MODE, name, kwargs, True,
result_chars=len(json.dumps(result, ensure_ascii=False, default=str)),
duration_ms=(time.time() - t0) * 1000)
return result
except Exception as e:
log_call(_TRANSPORT_MODE, name, kwargs, False, error=str(e),
duration_ms=(time.time() - t0) * 1000)
raise
return wrapper
return deco
_secret = os.environ.get("MCP_SECRET", "").strip().strip("/")
_http_path = f"/{_secret}/mcp" if _secret else "/mcp"
mcp = FastMCP(
"christian-corpus",
host=os.environ.get("HOST", "0.0.0.0"),
port=int(os.environ.get("PORT", "8000")),
streamable_http_path=_http_path,
stateless_http=True,
)
@mcp.tool()
@_logged("commentary")
def commentary(book: str, chapter: int, verse_start: int,
verse_end: Optional[int] = None,
query_text: Optional[str] = None,
max_sources: int = 4,
prefer: Optional[list[str]] = None,
max_chars: int = 8000) -> list[dict]:
"""Retrieve classic commentary excerpts covering a Bible verse or passage.
23 sources (Gill, Poole, Henry, Calvin, Ellicott, Alford, Meyer, JFB,
Clarke, Keil-Delitzsch, Robertson, Cambridge Greek, Spurgeon...), incl.
two cultural-background specialists: Lightfoot's Talmud & Hebraica
(rabbinic/Second-Temple context, Gospels) and Freeman's Manners & Customs
(whole Bible). Selection is relevance-ranked and perspective-diverse (MMR
over the question + one pick per style: devotional/critical/homiletical/
expository/background, narrower verse-ranges preferred). Pass query_text
(the user's actual
question) whenever available. prefer=[names] force-includes sources the
user asked for by name. max_chars bounds each returned excerpt.
"""
return _commentary.lookup_commentary(
book, chapter, verse_start, verse_end=verse_end, query_text=query_text,
max_sources=max_sources, prefer=prefer, max_chars=max_chars)
@mcp.tool()
@_logged("passage")
def passage(book: str, chapter: int, verse_start: int,
verse_end: Optional[int] = None, depth: str = "shallow") -> dict:
"""Read a Bible passage. depth='shallow': KJV text + Treasury of Scripture
Knowledge cross-references. depth='detailed': KJV/ASV/WEB/BSB translations
plus per-word Hebrew/Greek morphology with Strong's numbers and glosses.
"""
return _passage.lookup_passage(book, chapter, verse_start,
verse_end=verse_end, depth=depth)
@mcp.tool()
@_logged("interlinear")
def interlinear(book: str, chapter: int, verse_start: int,
verse_end: Optional[int] = None, edition: str = "SBL",
include_variants: bool = False,
expand_morphology: bool = True) -> dict:
"""Original-language text and per-token morphology for a verse or range.
Greek defaults to the coherent SBL token stream; edition may be SBL,
NA27, NA28, TH, WH, Treg, TR, Byz, NIV, ESV, or Tyn. Hebrew uses the
Masoretic text and prefers Qere while preserving Ketiv as an optional
variant. Each token includes surface form, lemma, Strong's number, full
morphology code, gloss, and (by default) an expanded morphology label.
With include_variants=True, the response also reports known whole-passage
inclusion/omission units in passage_variants. Their source and edition-level
caveat are returned explicitly; they are not a manuscript apparatus.
"""
return _interlinear.lookup_interlinear(
book, chapter, verse_start, verse_end=verse_end, edition=edition,
include_variants=include_variants,
expand_morphology=expand_morphology,
)
@mcp.tool()
@_logged("parse_morphology")
def parse_morphology(code: str, language: str = "") -> dict:
"""Expand a STEPBible Greek/Hebrew morphology code into readable grammar.
Examples: V-AAI-3S = Greek aorist active indicative third singular;
HVqi1cp = Hebrew Qal imperfect first common plural.
"""
return _interlinear.parse_morphology(code, language=language)
@mcp.tool()
@_logged("lexicon")
def lexicon(strongs: list[str]) -> dict:
"""Full lexicon entries for Strong's numbers (e.g. ["G26", "H430"];
zero-padding tolerated). Greek: Strong's + Thayer's + Liddell-Scott-Jones
+ Septuagint Hebrew equivalents. Hebrew: Strong's + Brown-Driver-Briggs +
Ancient Hebrew Lexicon (pictographic roots)."""
return _lexicon.lookup_lexicon(strongs)
@mcp.tool()
@_logged("topic")
def topic(keyword: str, max_headings: int = 5) -> list[dict]:
"""Topical Bible lookup: Nave's Topical Bible + Torrey's New Topical
Textbook (12,945 topics). Returns headings with subtopics and verse
references β€” the map of what Scripture says about a subject.
If no heading matches the keyword, returns a single entry with
heading=null and did_you_mean=[nearest headings]: pick the suggestion
that genuinely fits the theme and call topic() again with it, or
proceed without topical data if none fits."""
return _topic.lookup_topic(keyword, max_headings=max_headings)
@mcp.tool()
@_logged("encyclopedia")
def encyclopedia(term: str, max_results: int = 3, max_chars: int = 8000) -> list[dict]:
"""Bible encyclopedia articles about entities and subjects: people, places,
customs, institutions, doctrines, church history. Sources: International
Standard Bible Encyclopedia (1915) and McClintock & Strong's Cyclopedia
(49,767 articles incl. theology, denominations, biographies)."""
return _encyclopedia.lookup_encyclopedia(term, max_results=max_results,
max_chars=max_chars)
@mcp.tool()
@_logged("cross_references")
def cross_references(book: str, chapter: int, verse_start: int,
verse_end: Optional[int] = None, max_refs: int = 10) -> list[str]:
"""Treasury of Scripture Knowledge cross-references for a verse or range β€”
editor-vetted parallel passages, not model recall."""
return _tsk.get_cross_refs(book, chapter, verse_start,
chapter, verse_end or verse_start, max_refs=max_refs)
@mcp.tool()
@_logged("cited_by")
def cited_by(book: str, chapter: int, verse: int,
max_refs: int = 20) -> list[str]:
"""Reverse TSK lookup: verses whose cross-reference entries cite this verse."""
return _tsk.get_cited_by(book, chapter, verse, max_refs=max_refs)
@mcp.tool()
@_logged("cross_reference_paths")
def cross_reference_paths(book: str, chapter: int, verse: int,
depth: int = 2, max_nodes: int = 50,
max_edges: int = 200) -> dict:
"""Traverse TSK references up to four hops, ranked by independent paths."""
return _tsk.traverse_cross_refs(
book, chapter, verse, depth=depth,
max_nodes=max_nodes, max_edges=max_edges,
)
@mcp.tool()
@_logged("versification")
def versification(reference: str, from_tradition: str = "English",
to_tradition: str = "Hebrew") -> dict:
"""Map a reference among standard English, Hebrew/MT, Latin, and Greek/LXX numbering."""
return _versification.versification(reference, from_tradition, to_tradition)
@mcp.tool()
@_logged("phrase_occurrences")
def phrase_occurrences(strongs: list[str], adjacent: bool = True,
book_filter: str = "", edition: str = "SBL",
max_results: int = 50) -> dict:
"""Find adjacent or ordered Strong's-number sequences in the tagged text."""
return _original_search.phrase_occurrences(
strongs, adjacent=adjacent, book_filter=book_filter,
edition=edition, max_results=max_results,
)
@mcp.tool()
@_logged("search_original")
def search_original(query: str, field: str = "lemma", language: str = "",
book_filter: str = "", edition: str = "SBL",
match: str = "exact", max_results: int = 50) -> dict:
"""Search Greek/Hebrew pointed surface forms or lemmas, independent of English translation."""
return _original_search.search_original(
query, field=field, language=language, book_filter=book_filter,
edition=edition, match=match, max_results=max_results,
)
@mcp.tool()
@_logged("search_scripture")
def search_scripture(query: str, max_results: int = 8) -> list[dict]:
"""Keyword search (BM25) across all 31,102 KJV verses. Use when the user
describes an idea or wording without naming a specific passage --
"where does the Bible say..." Deterministic, no semantic guessing."""
return _search.search_scripture(query, max_results=max_results)
@mcp.tool()
@_logged("word_occurrences")
def word_occurrences(strongs: str, book_filter: Optional[str] = None,
max_results: int = 50) -> dict:
"""Every verse where a Strong's-numbered word occurs (e.g. "G26" = agape).
Returns total_occurrences (frequency, often used to judge a word's
importance) plus a sample of occurrences with immediate gloss."""
return _word_occ.word_occurrences(strongs, book_filter=book_filter, max_results=max_results)
@mcp.tool()
@_logged("historical_documents")
def historical_documents(query: str, max_results: int = 5) -> list[dict]:
"""Search the 17 historic creeds, confessions, and catechisms (Apostles',
Nicene, Westminster + catechisms, Heidelberg, Belgic, Dort, 1689 London
Baptist, Augsburg, 39 Articles, Chalcedon, Athanasian, plus Trent/
Dositheus/Baltimore for comparative reference) by topic β€” plus Mackie's
Bible Manners and Customs (1898) for daily-life cultural background."""
return _historical.historical_documents(query, max_results=max_results)
@mcp.tool()
@_logged("parallel_passages")
def parallel_passages(book: str, chapter: int, verse: int, max_results: int = 6) -> list[dict]:
"""Other passages paralleled with this one (UBS Parallel Passage
Database): synoptic Gospel parallels, and Old Testament verses quoted in
the New Testament (or vice versa). When a match crosses Old/New
Testament, its own text plus 1-2 commentary excerpts are included."""
return _parallel.parallel_passages(book, chapter, verse, max_results=max_results)
if _PRIVATE_MODE:
# Deliberately local/conditional and deliberately NOT wrapped in _logged:
# the public logger stores full arguments and must never receive private
# library queries, paths, excerpts, or identifiers.
from pathlib import Path
from private_rag.service import get_private_rag_service
_private = get_private_rag_service(repo_root=Path(__file__).resolve().parent)
@mcp.tool()
def library_health() -> dict:
"""Status and integrity counts for the opt-in local private evidence index."""
return _private.health()
@mcp.tool()
def library_sources(
collections: Optional[list[str]] = None,
include_direct_query_only: bool = False,
max_results: int = 100,
) -> dict:
"""List bounded evidence sources and authority classes without filesystem paths.
Scripture is not stored in the relational RAG and remains primary;
every listed source is subordinate.
"""
return _private.sources(
collections=collections,
include_direct_query_only=include_direct_query_only,
max_results=max_results,
)
@mcp.tool()
def library_search(
query: str,
concepts: Optional[list[str]] = None,
collections: Optional[list[str]] = None,
authors: Optional[list[str]] = None,
source_ids: Optional[list[str]] = None,
passages: Optional[list[str]] = None,
mode: str = "hybrid",
max_results: int = 8,
max_chars: int = 1400,
use_reranker: bool = True,
) -> dict:
"""Retrieve cited research evidence; it does not interpret Scripture or teach.
Scope: a 30-source relational-research index (friendship, attachment, relationship science, honor-shame and NT social world, Gospel narrative method, Greek verbal aspect, Jesus' emotions). It holds no general theology library; off-topic queries return status insufficient_evidence because results below a reranker relevance floor are dropped.
Devotional and personal-study sources are excluded unless source_ids
names them or a passages filter requests mapped article notes. Broader
passage ranges match contained core annotations; an empty mapped search receives a marked
broad hybrid fallback followed by bounded section-level library context.
Common concept
aliases (for example agency, temptation, poverty, prayer, Gethsemane)
are accepted. Results retain population, applicability, and explicit
authority metadata; RAG evidence remains subordinate to Scripture.
"""
return _private.search(
query,
concepts=concepts,
collections=collections,
authors=authors,
source_ids=source_ids,
passages=passages,
mode=mode,
max_results=max_results,
max_chars=max_chars,
use_reranker=use_reranker,
)
@mcp.tool()
def library_context(
chunk_id: str,
scope: str = "section",
max_chars: int = 4000,
) -> dict:
"""Expand a result to its source-faithful context, retaining authority metadata."""
return _private.context(chunk_id, scope=scope, max_chars=max_chars)
if __name__ == "__main__":
if _PRIVATE_MODE and "--stdio" not in sys.argv:
raise SystemExit("Private RAG tools are local stdio-only; add --stdio")
if "--stdio" in sys.argv:
# Loading torch/sentence-transformers for the first time inside an
# active stdio request can stall the protocol on Windows. Warm the
# one shared model before FastMCP starts, and keep third-party output
# off stdout because stdout is the JSON-RPC wire in stdio mode.
from contextlib import redirect_stdout
from embedding_model import preload_embedder
_TRANSPORT_MODE = "stdio"
with redirect_stdout(sys.stderr):
preload_embedder()
if _PRIVATE_MODE:
_private.models.preload(include_reranker=True)
mcp.run()
else:
print(f"Streamable HTTP endpoint: http://{mcp.settings.host}:"
f"{mcp.settings.port}{_http_path}", flush=True)
mcp.run(transport="streamable-http")