Skip to content

Refactor MCP server for improved testability #8

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
polinom merged 1 commit into from
Jan 10, 2026
Merged

Refactor MCP server for improved testability #8

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 37 additions & 10 deletions telecode/mcp_server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,12 +86,8 @@ def _truncate_message(text: str, limit: int = 3500) -> str:
return f"{text[:limit]}\n...[truncated]"


# Initialize FastMCP server
mcp = FastMCP("Telecode MCP Server")


@mcp.tool()
def local_claude_code(
# Implementation functions (for testing and reuse)
def _local_claude_code_impl(
prompt: str,
session_id: Optional[str] = None,
timeout_s: Optional[int] = None,
Expand Down Expand Up @@ -138,8 +134,7 @@ def local_claude_code(
return f"Unexpected error: {type(exc).__name__}: {exc}"


@mcp.tool()
def local_codex(
def _local_codex_impl(
prompt: str,
session_id: Optional[str] = None,
timeout_s: Optional[int] = None,
Expand Down Expand Up @@ -199,8 +194,7 @@ def local_codex(
}


@mcp.tool()
def local_cli(command: str, timeout_s: int = 30) -> str:
def _local_cli_impl(command: str, timeout_s: int = 30) -> str:
"""Execute a shell command locally.

Security Warning:
Expand Down Expand Up @@ -240,6 +234,39 @@ def local_cli(command: str, timeout_s: int = 30) -> str:
return f"Error: Command failed: {exc}"


# Initialize FastMCP server
mcp = FastMCP("Telecode MCP Server")


# MCP tool wrappers (delegate to implementation functions)
@mcp.tool()
def local_claude_code(
prompt: str,
session_id: Optional[str] = None,
timeout_s: Optional[int] = None,
image_paths: Optional[list[str]] = None
) -> str:
"""Execute a prompt using Claude Code CLI."""
return _local_claude_code_impl(prompt, session_id, timeout_s, image_paths)


@mcp.tool()
def local_codex(
prompt: str,
session_id: Optional[str] = None,
timeout_s: Optional[int] = None,
image_paths: Optional[list[str]] = None
) -> dict:
"""Execute a prompt using Codex CLI."""
return _local_codex_impl(prompt, session_id, timeout_s, image_paths)


@mcp.tool()
def local_cli(command: str, timeout_s: int = 30) -> str:
"""Execute a shell command locally."""
return _local_cli_impl(command, timeout_s)


def create_mcp_app():
"""Create and configure MCP server app.

Expand Down
44 changes: 22 additions & 22 deletions tests/test_mcp_server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,8 +79,8 @@ def mock_run(cmd, **kwargs):

monkeypatch.setattr("subprocess.run", mock_run)

from telecode.mcp_server import local_cli
result = local_cli("echo test")
from telecode.mcp_server import _local_cli_impl
result = _local_cli_impl("echo test")
assert "test output" in result


Expand All @@ -91,8 +91,8 @@ def mock_run(cmd, **kwargs):

monkeypatch.setattr("subprocess.run", mock_run)

from telecode.mcp_server import local_cli
result = local_cli("sleep 100", timeout_s=10)
from telecode.mcp_server import _local_cli_impl
result = _local_cli_impl("sleep 100", timeout_s=10)
assert "timed out" in result.lower()


Expand All @@ -103,8 +103,8 @@ def mock_run(cmd, **kwargs):

monkeypatch.setattr("subprocess.run", mock_run)

from telecode.mcp_server import local_cli
result = local_cli("invalid_command")
from telecode.mcp_server import _local_cli_impl
result = _local_cli_impl("invalid_command")
assert "error" in result.lower() or "failed" in result.lower()


Expand All @@ -115,10 +115,10 @@ def mock_ask_claude_code(prompt, session_id, timeout_s, image_paths):

monkeypatch.setattr("telecode.mcp_server.ask_claude_code", mock_ask_claude_code)

from telecode.mcp_server import local_claude_code, _MCP_SESSIONS
from telecode.mcp_server import _local_claude_code_impl, _MCP_SESSIONS
_MCP_SESSIONS.clear()

result = local_claude_code("test prompt")
result = _local_claude_code_impl("test prompt")
assert result == "Mocked Claude response"


Expand All @@ -130,8 +130,8 @@ def mock_ask_claude_code(prompt, session_id, timeout_s, image_paths):

monkeypatch.setattr("telecode.mcp_server.ask_claude_code", mock_ask_claude_code)

from telecode.mcp_server import local_claude_code
result = local_claude_code("test prompt", session_id="my-session-123")
from telecode.mcp_server import _local_claude_code_impl
result = _local_claude_code_impl("test prompt", session_id="my-session-123")
assert result == "Mocked Claude response"


Expand All @@ -142,8 +142,8 @@ def mock_ask_claude_code(*args, **kwargs):

monkeypatch.setattr("telecode.mcp_server.ask_claude_code", mock_ask_claude_code)

from telecode.mcp_server import local_claude_code
result = local_claude_code("test", timeout_s=10)
from telecode.mcp_server import _local_claude_code_impl
result = _local_claude_code_impl("test", timeout_s=10)
assert "timed out" in result.lower()


Expand All @@ -154,8 +154,8 @@ def mock_ask_claude_code(*args, **kwargs):

monkeypatch.setattr("telecode.mcp_server.ask_claude_code", mock_ask_claude_code)

from telecode.mcp_server import local_claude_code
result = local_claude_code("test")
from telecode.mcp_server import _local_claude_code_impl
result = _local_claude_code_impl("test")
assert "error" in result.lower()
assert "session not found" in result.lower()

Expand All @@ -167,10 +167,10 @@ def mock_ask_codex_exec(prompt, session_id, timeout_s, image_paths):

monkeypatch.setattr("telecode.mcp_server.ask_codex_exec", mock_ask_codex_exec)

from telecode.mcp_server import local_codex, _MCP_SESSIONS
from telecode.mcp_server import _local_codex_impl, _MCP_SESSIONS
_MCP_SESSIONS.clear()

result = local_codex("test prompt")
result = _local_codex_impl("test prompt")
assert isinstance(result, dict)
assert result["answer"] == "Mocked answer"
assert result["session_id"] == "session-123"
Expand All @@ -185,8 +185,8 @@ def mock_ask_codex_exec(prompt, session_id, timeout_s, image_paths):

monkeypatch.setattr("telecode.mcp_server.ask_codex_exec", mock_ask_codex_exec)

from telecode.mcp_server import local_codex
result = local_codex("test prompt", session_id="my-session-456")
from telecode.mcp_server import _local_codex_impl
result = _local_codex_impl("test prompt", session_id="my-session-456")
assert result["session_id"] == "my-session-456"


Expand All @@ -197,8 +197,8 @@ def mock_ask_codex_exec(*args, **kwargs):

monkeypatch.setattr("telecode.mcp_server.ask_codex_exec", mock_ask_codex_exec)

from telecode.mcp_server import local_codex
result = local_codex("test", timeout_s=10)
from telecode.mcp_server import _local_codex_impl
result = _local_codex_impl("test", timeout_s=10)
assert isinstance(result, dict)
assert "timed out" in result["answer"].lower()

Expand All @@ -210,8 +210,8 @@ def mock_ask_codex_exec(*args, **kwargs):

monkeypatch.setattr("telecode.mcp_server.ask_codex_exec", mock_ask_codex_exec)

from telecode.mcp_server import local_codex
result = local_codex("test")
from telecode.mcp_server import _local_codex_impl
result = _local_codex_impl("test")
assert isinstance(result, dict)
assert "error" in result["answer"].lower()
assert "codex failed" in result["answer"].lower()
Expand Down