diff --git a/mcp_server.py b/mcp_server.py index a87feba..f6e5a4d 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -29,7 +29,7 @@ _REPO = Path(__file__).parent sys.path.insert(0, str(_REPO)) -from mcp.server.fastmcp import FastMCP +from mcp.server.mcpserver import MCPServer from query.dispatch import query_file from query.config import Root, load_config, normalize_path @@ -276,7 +276,7 @@ def _truncate(output: str) -> tuple[str, bool]: # -- MCP server ---------------------------------------------------------------- -mcp = FastMCP("tscodesearch") +mcp = MCPServer("tscodesearch") # -- query_codebase ------------------------------------------------------------ diff --git a/tests/unit/test_mcp_server.py b/tests/unit/test_mcp_server.py index 759a60a..d99a20f 100644 --- a/tests/unit/test_mcp_server.py +++ b/tests/unit/test_mcp_server.py @@ -7,6 +7,8 @@ from __future__ import annotations import sys +import importlib.util +import subprocess import tempfile import time import unittest @@ -21,22 +23,22 @@ # -- Stub mcp if not installed (indexserver venv lacks it) -------------------- -if "mcp" not in sys.modules: +if importlib.util.find_spec("mcp") is None: import types from unittest.mock import MagicMock - class _FakeFastMCP: + class _FakeMCPServer: def __init__(self, *_a, **_kw): pass def tool(self): def deco(fn): return fn return deco def run(self): pass - _fastmcp_mod = types.ModuleType("mcp.server.fastmcp") - setattr(_fastmcp_mod, "FastMCP", _FakeFastMCP) + _mcpserver_mod = types.ModuleType("mcp.server.mcpserver") + setattr(_mcpserver_mod, "MCPServer", _FakeMCPServer) sys.modules.setdefault("mcp", MagicMock()) sys.modules.setdefault("mcp.server", MagicMock()) - sys.modules["mcp.server.fastmcp"] = _fastmcp_mod + sys.modules["mcp.server.mcpserver"] = _mcpserver_mod # -- Import mcp_server, skip all tests if config.json is absent --------------- @@ -54,6 +56,21 @@ def run(self): pass _skip = unittest.skipUnless(_IMPORT_OK, f"mcp_server import failed: {_IMPORT_ERR}") +class TestMCPServerRuntime(unittest.TestCase): + + @unittest.skipIf(importlib.util.find_spec("mcp") is None, "mcp package is not installed") + def test_real_server_starts_and_handles_eof(self): + result = subprocess.run( + [sys.executable, str(REPO_ROOT / "mcp_server.py")], + input="", + text=True, + capture_output=True, + timeout=15, + check=False, + ) + assert result.returncode == 0, result.stderr + + @_skip class TestLoggedTool(unittest.TestCase):