user2595/codex-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mcp): FastMCP server exposing read-only KB tools (stdio)

Browse Source 
Implements F-14: thin FastMCP wrapper over existing codex domain modules.
Seven read-only tools: search, ask, wiki_read, wiki_list, discover_leads,
provenance_verify, synthesis_browse. All optional-feature tools degrade
gracefully to {"error": "feature not available"} instead of crashing.
Adds mcp[cli]>=1.0 dependency and codex-mcp console_script entry-point.
28 new tests across test_server, test_search, test_readonly, test_graceful,
test_http_auth — all green; 0 regressions in existing 172 tests.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Tarik Moussa
2026-06-14 08:23:22 +02:00
parent 59763105a5
commit b803811701
10 changed files with 1484 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
tests/mcp/__init__.py Normal file
View File

211
tests/mcp/test_graceful.py Normal file
View File

@@ -0,0 +1,211 @@
"""test_graceful.py — missing/unavailable features must degrade, never crash.
All tools that wrap optional features (wiki, synthesis leads, provenance)
must return a structured "feature not available" response instead of raising.
"""
from __future__ import annotations
import json
import tempfile
from pathlib import Path
from unittest.mock import patch
# ---------------------------------------------------------------------------
# ask — always returns "not available" (RAG not implemented)
# ---------------------------------------------------------------------------
def test_ask_returns_not_available() -> None:
"""ask() must return {'error': 'not available'} — RAG not yet implemented."""
from codex.mcp_server import ask
result = ask("What is conformal geometry?")
assert result == {"error": "not available"}
# ---------------------------------------------------------------------------
# wiki_read — page missing → "feature not available"
# ---------------------------------------------------------------------------
def test_wiki_read_missing_page_returns_not_available() -> None:
"""wiki_read() with a nonexistent page must return {"error": "feature not available"}."""
from codex.mcp_server import wiki_read
with (
tempfile.TemporaryDirectory() as tmpdir,
patch("codex.config.get_settings") as mock_cfg,
):
mock_cfg.return_value.wiki_dir = tmpdir
result = wiki_read("nonexistent_concept")
assert result == {"error": "feature not available"}
def test_wiki_read_existing_page_returns_markdown() -> None:
"""wiki_read() with an existing page must return {markdown, sources}."""
from codex.mcp_server import wiki_read
with (
tempfile.TemporaryDirectory() as tmpdir,
patch("codex.config.get_settings") as mock_cfg,
):
mock_cfg.return_value.wiki_dir = tmpdir
# Write a fake wiki page
page = Path(tmpdir) / "conformal_maps.md"
page.write_text("# Conformal Maps\n\nSome content.", encoding="utf-8")
result = wiki_read("conformal_maps")
assert "markdown" in result
assert "sources" in result
assert "# Conformal Maps" in str(result["markdown"])
def test_wiki_read_exception_returns_not_available() -> None:
"""wiki_read() must degrade gracefully when get_settings raises."""
from codex.mcp_server import wiki_read
with patch("codex.config.get_settings", side_effect=RuntimeError("config error")):
result = wiki_read("any_slug")
assert result == {"error": "feature not available"}
# ---------------------------------------------------------------------------
# wiki_list — missing wiki dir → empty list
# ---------------------------------------------------------------------------
def test_wiki_list_missing_dir_returns_empty() -> None:
"""wiki_list() must return [] when the wiki directory does not exist."""
from codex.mcp_server import wiki_list
with (
tempfile.TemporaryDirectory() as tmpdir,
patch("codex.config.get_settings") as mock_cfg,
):
mock_cfg.return_value.wiki_dir = str(Path(tmpdir) / "no_wiki_here")
result = wiki_list()
assert result == []
def test_wiki_list_existing_dir_returns_pages() -> None:
"""wiki_list() must return page entries for .md files in the wiki dir."""
from codex.mcp_server import wiki_list
with (
tempfile.TemporaryDirectory() as tmpdir,
patch("codex.config.get_settings") as mock_cfg,
):
mock_cfg.return_value.wiki_dir = tmpdir
(Path(tmpdir) / "concept_a.md").write_text("# A", encoding="utf-8")
(Path(tmpdir) / "concept_b.md").write_text("# B", encoding="utf-8")
(Path(tmpdir) / "index.md").write_text("# Index", encoding="utf-8") # must be skipped
result = wiki_list()
slugs = {r["slug"] for r in result}
assert "concept_a" in slugs
assert "concept_b" in slugs
assert "index" not in slugs # index.md is excluded
# ---------------------------------------------------------------------------
# discover_leads — domain function raises → error dict, no crash
# ---------------------------------------------------------------------------
def test_discover_leads_db_error_returns_error() -> None:
"""discover_leads() must return [{"error": ...}] when the discovery call fails."""
from codex.mcp_server import discover_leads
# Patch the domain function directly so the exception always surfaces
with patch("codex.discover.discovery_leads", side_effect=RuntimeError("no DB")):
result = discover_leads(limit=5)
assert len(result) == 1
assert "error" in result[0]
# ---------------------------------------------------------------------------
# synthesis_browse — missing leads dir → "feature not available"
# ---------------------------------------------------------------------------
def test_synthesis_browse_missing_dir_returns_not_available() -> None:
"""synthesis_browse() must return [{'error': 'feature not available'}] when leads/ absent."""
from codex.mcp_server import synthesis_browse
with (
tempfile.TemporaryDirectory() as tmpdir,
patch("codex.config.get_settings") as mock_cfg,
):
mock_cfg.return_value.leads_dir = str(Path(tmpdir) / "no_leads_here")
result = synthesis_browse()
assert result == [{"error": "feature not available"}]
def test_synthesis_browse_existing_dir_returns_items() -> None:
"""synthesis_browse() must parse and return JSON files from the leads dir."""
from codex.mcp_server import synthesis_browse
with (
tempfile.TemporaryDirectory() as tmpdir,
patch("codex.config.get_settings") as mock_cfg,
):
leads_dir = Path(tmpdir) / "leads"
leads_dir.mkdir()
(leads_dir / "lead1.json").write_text(
json.dumps({"kind": "connection", "title": "C1"}), encoding="utf-8"
)
(leads_dir / "lead2.json").write_text(
json.dumps({"kind": "conjecture", "title": "C2"}), encoding="utf-8"
)
mock_cfg.return_value.leads_dir = str(leads_dir)
result = synthesis_browse(kind="all")
assert len(result) == 2
kinds = {str(r.get("kind")) for r in result}
assert "connection" in kinds
assert "conjecture" in kinds
def test_synthesis_browse_kind_filter() -> None:
"""synthesis_browse(kind='conjecture') must return only conjectures."""
from codex.mcp_server import synthesis_browse
with (
tempfile.TemporaryDirectory() as tmpdir,
patch("codex.config.get_settings") as mock_cfg,
):
leads_dir = Path(tmpdir) / "leads"
leads_dir.mkdir()
(leads_dir / "lead1.json").write_text(
json.dumps({"kind": "connection", "title": "C1"}), encoding="utf-8"
)
(leads_dir / "lead2.json").write_text(
json.dumps({"kind": "conjecture", "title": "C2"}), encoding="utf-8"
)
mock_cfg.return_value.leads_dir = str(leads_dir)
result = synthesis_browse(kind="conjecture")
assert all(str(r.get("kind")) == "conjecture" for r in result)
assert len(result) == 1
# ---------------------------------------------------------------------------
# provenance_verify — exception → "feature not available"
# ---------------------------------------------------------------------------
def test_provenance_verify_exception_returns_not_available() -> None:
"""provenance_verify() must return [{"error": ...}] on scan failure."""
from codex.mcp_server import provenance_verify
with patch("codex.provenance.scan_cite_tags", side_effect=RuntimeError("scan failed")):
result = provenance_verify("/some/nonexistent/path")
assert len(result) == 1
assert "error" in result[0]

121
tests/mcp/test_http_auth.py Normal file
View File

@@ -0,0 +1,121 @@
"""test_http_auth.py — HTTP transport token-auth enforcement.
Tests:
1. _require_http_token() raises RuntimeError when MCP_AUTH_TOKEN is absent.
2. _require_http_token() returns the token when it is set.
3. main() with transport='http' raises without token (refuses to start).
4. main() with transport='stdio' runs mcp.run (happy path, no token needed).
"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
# ---------------------------------------------------------------------------
# _require_http_token
# ---------------------------------------------------------------------------
def test_require_http_token_raises_without_token() -> None:
"""_require_http_token() must raise RuntimeError when mcp_auth_token is None."""
from codex.mcp_server import _require_http_token
mock_settings = MagicMock()
mock_settings.mcp_auth_token = None
with (
patch("codex.config.get_settings", return_value=mock_settings),
pytest.raises(RuntimeError, match="MCP_AUTH_TOKEN"),
):
_require_http_token()
def test_require_http_token_returns_token_when_set() -> None:
"""_require_http_token() must return the token string when it is configured."""
from codex.mcp_server import _require_http_token
mock_settings = MagicMock()
mock_settings.mcp_auth_token = "super-secret-token"
with patch("codex.config.get_settings", return_value=mock_settings):
result = _require_http_token()
assert result == "super-secret-token"
# ---------------------------------------------------------------------------
# main() — HTTP transport without token → RuntimeError
# ---------------------------------------------------------------------------
def test_main_http_without_token_raises() -> None:
"""main() with MCP_TRANSPORT=http and no token must raise RuntimeError."""
from codex.mcp_server import main
mock_settings = MagicMock()
mock_settings.mcp_transport = "http"
mock_settings.mcp_auth_token = None
with (
patch("codex.config.get_settings", return_value=mock_settings),
pytest.raises(RuntimeError, match="MCP_AUTH_TOKEN"),
):
main()
# ---------------------------------------------------------------------------
# main() — stdio transport runs mcp.run
# ---------------------------------------------------------------------------
def test_main_stdio_calls_mcp_run() -> None:
"""main() with stdio transport must call mcp.run(transport='stdio')."""
from codex import mcp_server
mock_settings = MagicMock()
mock_settings.mcp_transport = "stdio"
with (
patch("codex.config.get_settings", return_value=mock_settings),
patch.object(mcp_server.mcp, "run") as mock_run,
):
mcp_server.main()
mock_run.assert_called_once_with(transport="stdio")
# ---------------------------------------------------------------------------
# main() — HTTP transport with valid token attempts uvicorn.run
# ---------------------------------------------------------------------------
def test_main_http_with_token_calls_uvicorn() -> None:
"""main() with http transport + valid token must invoke uvicorn.run."""
from codex import mcp_server
mock_settings = MagicMock()
mock_settings.mcp_transport = "http"
mock_settings.mcp_auth_token = "test-token"
mock_settings.mcp_host = "127.0.0.1"
mock_settings.mcp_port = 8765
# Stub the ASGI app returned by streamable_http_app
mock_app = MagicMock()
mock_app.add_middleware = MagicMock()
with (
patch("codex.config.get_settings", return_value=mock_settings),
patch.object(mcp_server.mcp, "streamable_http_app", return_value=mock_app),
patch("uvicorn.run") as mock_uvicorn,
):
mcp_server.main()
mock_uvicorn.assert_called_once()
call_kwargs = mock_uvicorn.call_args
assert call_kwargs is not None
# host and port must match settings
_, kwargs = call_kwargs
assert kwargs.get("host") == "127.0.0.1"
assert kwargs.get("port") == 8765

83
tests/mcp/test_readonly.py Normal file
View File

@@ -0,0 +1,83 @@
"""test_readonly.py — CRITICAL: ensure no write/ingest tools are exposed via MCP.
This test MUST fail if any write or ingest tool is accidentally registered.
It is the primary enforcement point for the read-only invariant of F-14.
Read-only tools (allowed):
search, ask, wiki_read, wiki_list, discover_leads,
provenance_verify, synthesis_browse
Write/ingest tools that must NEVER appear:
Any tool whose name contains an ingest/write/delete/add/update/remove
verb, or any of the known write-mode CLI operations.
"""
from __future__ import annotations
import asyncio
import re
# ---------------------------------------------------------------------------
# Write-verb pattern — any tool matching this is a forbidden write tool.
# ---------------------------------------------------------------------------
_WRITE_VERB_RE = re.compile(
r"(?:^|_)(ingest|write|delete|remove|add|update|upsert|insert|create|sync|compile|export)",
re.IGNORECASE,
)
# Explicit allowlist — only these names may appear.
ALLOWED_TOOLS = frozenset(
{
"search",
"ask",
"wiki_read",
"wiki_list",
"discover_leads",
"provenance_verify",
"synthesis_browse",
}
)
def test_no_write_tools_registered() -> None:
"""CRITICAL: fail if any write-verb tool is registered on the MCP server."""
from codex.mcp_server import mcp
tools = asyncio.run(mcp.list_tools())
registered = {t.name for t in tools}
violations: list[str] = []
for name in registered:
if _WRITE_VERB_RE.search(name):
violations.append(name)
assert not violations, (
f"Write/ingest tools found in MCP server (FORBIDDEN): {violations}. "
"MCP server must expose read-only tools only."
)
def test_only_allowed_tools_registered() -> None:
"""CRITICAL: registered tools must be a subset of the explicit allowlist."""
from codex.mcp_server import mcp
tools = asyncio.run(mcp.list_tools())
registered = {t.name for t in tools}
disallowed = registered - ALLOWED_TOOLS
assert not disallowed, (
f"Tools not on the read-only allowlist: {disallowed}. "
"Add them to ALLOWED_TOOLS only after verifying they are read-only."
)
def test_all_allowed_tools_present() -> None:
"""Every tool in the allowlist must be registered (completeness check)."""
from codex.mcp_server import mcp
tools = asyncio.run(mcp.list_tools())
registered = {t.name for t in tools}
missing = ALLOWED_TOOLS - registered
assert not missing, f"Expected read-only tools missing from server: {missing}"

176
tests/mcp/test_search.py Normal file
View File

@@ -0,0 +1,176 @@
"""test_search.py — verify the search tool: retrieval, return format, error handling.
Since search() imports get_embedder and get_conn lazily (inside the function body),
we patch them at their source modules: codex.embed and codex.db.
"""
from __future__ import annotations
from contextlib import contextmanager
from typing import Any
from unittest.mock import MagicMock, patch
import numpy as np
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_conn_cm(mock_conn: MagicMock) -> Any:
"""Return a context-manager factory that yields *mock_conn*."""
@contextmanager # type: ignore[arg-type]
def _cm() -> Any:
yield mock_conn
return _cm
def _fake_row(
bibkey: str,
paper_id: str,
ord_val: int,
content: str,
dist: float,
) -> dict[str, Any]:
return {
"bibkey": bibkey,
"paper_id": paper_id,
"ord": ord_val,
"content": content,
"dist": dist,
}
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
def test_search_returns_correct_keys() -> None:
"""search() must return dicts with the required five keys."""
from codex.mcp_server import search
mock_conn = MagicMock()
mock_conn.execute.return_value.fetchall.return_value = [
_fake_row("Springborn2008", "springborn-2008", 16, "The volume formula is V = ...", 0.2),
]
fake_dense = np.zeros((1, 1024), dtype=np.float32)
mock_embedder = MagicMock()
mock_embedder.encode_dense.return_value = fake_dense
with (
patch("codex.embed.get_embedder", return_value=mock_embedder),
patch("codex.db.get_conn", side_effect=_make_conn_cm(mock_conn)),
):
results = search("volume formula", limit=5)
assert len(results) == 1
hit = results[0]
assert set(hit.keys()) == {"bibkey", "paper_id", "locator", "score", "snippet"}
def test_search_score_is_similarity_not_distance() -> None:
"""score = 1 - distance, so a dist of 0.0 should yield score 1.0."""
from codex.mcp_server import search
mock_conn = MagicMock()
mock_conn.execute.return_value.fetchall.return_value = [
_fake_row("Author2020", "paper-x", 3, "Some chunk content here.", 0.0),
]
fake_dense = np.zeros((1, 1024), dtype=np.float32)
mock_embedder = MagicMock()
mock_embedder.encode_dense.return_value = fake_dense
with (
patch("codex.embed.get_embedder", return_value=mock_embedder),
patch("codex.db.get_conn", side_effect=_make_conn_cm(mock_conn)),
):
results = search("some query")
assert results[0]["score"] == 1.0
def test_search_locator_format() -> None:
"""locator must be 'chunk <ord>'."""
from codex.mcp_server import search
mock_conn = MagicMock()
mock_conn.execute.return_value.fetchall.return_value = [
_fake_row("Auth2021", "paper-y", 42, "Content.", 0.3),
]
fake_dense = np.zeros((1, 1024), dtype=np.float32)
mock_embedder = MagicMock()
mock_embedder.encode_dense.return_value = fake_dense
with (
patch("codex.embed.get_embedder", return_value=mock_embedder),
patch("codex.db.get_conn", side_effect=_make_conn_cm(mock_conn)),
):
results = search("query")
assert results[0]["locator"] == "chunk 42"
def test_search_snippet_truncated_to_300() -> None:
"""snippet must be truncated to 300 characters."""
from codex.mcp_server import search
long_content = "x" * 500
mock_conn = MagicMock()
mock_conn.execute.return_value.fetchall.return_value = [
_fake_row("Auth2022", "paper-z", 1, long_content, 0.1),
]
fake_dense = np.zeros((1, 1024), dtype=np.float32)
mock_embedder = MagicMock()
mock_embedder.encode_dense.return_value = fake_dense
with (
patch("codex.embed.get_embedder", return_value=mock_embedder),
patch("codex.db.get_conn", side_effect=_make_conn_cm(mock_conn)),
):
results = search("query")
assert len(results[0]["snippet"]) == 300 # type: ignore[arg-type]
def test_search_returns_error_dict_on_exception() -> None:
"""When the embedder raises, search must return [{"error": ...}], not crash."""
from codex.mcp_server import search
mock_embedder = MagicMock()
mock_embedder.encode_dense.side_effect = RuntimeError("DB down")
with patch("codex.embed.get_embedder", return_value=mock_embedder):
results = search("query")
assert len(results) == 1
assert "error" in results[0]
def test_search_provenance_keys() -> None:
"""Each hit must carry bibkey and paper_id for provenance traceability."""
from codex.mcp_server import search
mock_conn = MagicMock()
mock_conn.execute.return_value.fetchall.return_value = [
_fake_row("Crane1999", "crane-1999", 7, "Discrete exterior calculus ...", 0.15),
]
fake_dense = np.zeros((1, 1024), dtype=np.float32)
mock_embedder = MagicMock()
mock_embedder.encode_dense.return_value = fake_dense
with (
patch("codex.embed.get_embedder", return_value=mock_embedder),
patch("codex.db.get_conn", side_effect=_make_conn_cm(mock_conn)),
):
results = search("exterior calculus")
assert results[0]["bibkey"] == "Crane1999"
assert results[0]["paper_id"] == "crane-1999"

49
tests/mcp/test_server.py Normal file
View File

@@ -0,0 +1,49 @@
"""test_server.py — verify the MCP server instantiates and all tools are registered."""
from __future__ import annotations
import asyncio
# ---------------------------------------------------------------------------
# Expected tools (must match EXACTLY what mcp_server.py registers)
# ---------------------------------------------------------------------------
EXPECTED_TOOLS = {
"search",
"ask",
"wiki_read",
"wiki_list",
"discover_leads",
"provenance_verify",
"synthesis_browse",
}
def test_server_instance_exists() -> None:
"""The module-level ``mcp`` FastMCP instance must exist."""
from codex.mcp_server import mcp
assert mcp is not None
assert mcp.name == "codex"
def test_all_tools_registered() -> None:
"""Every expected tool name must be registered on the server."""
from codex.mcp_server import mcp
tools = asyncio.run(mcp.list_tools())
registered = {t.name for t in tools}
missing = EXPECTED_TOOLS - registered
assert not missing, f"Tools not registered: {missing}"
def test_no_extra_unexpected_write_tools() -> None:
"""No tool outside the expected set should be registered (belt-and-suspenders)."""
from codex.mcp_server import mcp
tools = asyncio.run(mcp.list_tools())
registered = {t.name for t in tools}
unexpected = registered - EXPECTED_TOOLS
assert not unexpected, f"Unexpected tools registered: {unexpected}"