homelab-mcp
0212913831
MCP server exposing self-hosted Gitea (gitlab.impresion3d.pro) via stdio JSON-RPC 2.0. Six tools: list_repos, get_repo, get_file, create_issue, list_open_prs, merge_pr. The merge_pr tool pins the Gitea 'do'-field trap (see skill gitea pitfall #13) — sends lowercase 'do', treats HTTP 200 with non-empty body as failure. Stack: FastMCP + httpx + pydantic + uv, with respx-mocked pytest suite (15 tests, 0.69s) and ruff + mypy strict green. Verification: scripts/verify_stdio.py boots the server as a subprocess, exchanges real MCP messages over stdio, and calls list_repos against the live Gitea instance — all green.
122 lines
3.9 KiB
Python
122 lines
3.9 KiB
Python
"""End-to-end stdio verification.
|
|
|
|
Boots the MCP server as a subprocess, exchanges real JSON-RPC 2.0 messages
|
|
over stdio (the same transport Claude Desktop / mcp-cli use), and verifies:
|
|
|
|
1. The initialize handshake completes and reports our 6 tools.
|
|
2. tools/list returns the same 6 tools with proper JSON Schema.
|
|
3. tools/call (list_repos) hits the real Gitea instance and returns data.
|
|
|
|
This is the closest you can get to "what does a real MCP client see?"
|
|
without installing one. If this script passes, the server speaks the
|
|
protocol correctly.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def run_e2e() -> None:
|
|
env = os.environ.copy()
|
|
if "GITEA_URL" not in env or "GITEA_TOKEN" not in env:
|
|
print("ERROR: GITEA_URL and GITEA_TOKEN must be in env.", file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
cmd = ["uv", "run", "homelab-mcp-gitea"]
|
|
proc = subprocess.Popen(
|
|
cmd,
|
|
cwd=Path(__file__).parent.parent,
|
|
env=env,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
|
|
def send(msg: dict[str, Any]) -> dict[str, Any]:
|
|
line = json.dumps(msg) + "\n"
|
|
assert proc.stdin is not None
|
|
proc.stdin.write(line)
|
|
proc.stdin.flush()
|
|
assert proc.stdout is not None
|
|
response_line = proc.stdout.readline()
|
|
return json.loads(response_line)
|
|
|
|
try:
|
|
# 1) initialize
|
|
init_resp = send(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "initialize",
|
|
"params": {
|
|
"protocolVersion": "2024-11-05",
|
|
"capabilities": {},
|
|
"clientInfo": {"name": "stdio-verifier", "version": "0.1.0"},
|
|
},
|
|
}
|
|
)
|
|
assert "result" in init_resp, f"initialize failed: {init_resp}"
|
|
server_info = init_resp["result"]["serverInfo"]
|
|
print(f"server: {server_info['name']} v{server_info['version']}")
|
|
|
|
# MCP requires the initialized notification before tool calls
|
|
assert proc.stdin is not None
|
|
proc.stdin.write(
|
|
json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"}) + "\n"
|
|
)
|
|
proc.stdin.flush()
|
|
|
|
# 2) tools/list
|
|
tools_resp = send({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}})
|
|
assert "result" in tools_resp, f"tools/list failed: {tools_resp}"
|
|
tools = tools_resp["result"]["tools"]
|
|
tool_names = sorted(t["name"] for t in tools)
|
|
print(f"tools registered ({len(tools)}): {tool_names}")
|
|
expected = {
|
|
"create_issue",
|
|
"get_file",
|
|
"get_repo",
|
|
"list_open_prs",
|
|
"list_repos",
|
|
"merge_pr",
|
|
}
|
|
assert set(tool_names) == expected, f"unexpected tools: {tool_names}"
|
|
|
|
# 3) tools/call list_repos
|
|
call_resp = send(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": 3,
|
|
"method": "tools/call",
|
|
"params": {"name": "list_repos", "arguments": {"limit": 3}},
|
|
}
|
|
)
|
|
assert "result" in call_resp, f"tools/call failed: {call_resp}"
|
|
content = call_resp["result"]["content"]
|
|
assert content and content[0]["type"] == "text"
|
|
text = content[0]["text"]
|
|
print("list_repos returned:")
|
|
for line in text.splitlines()[:5]:
|
|
print(f" {line}")
|
|
|
|
print("\nALL CHECKS PASSED — MCP server speaks the protocol correctly.")
|
|
finally:
|
|
proc.terminate()
|
|
try:
|
|
proc.wait(timeout=3)
|
|
except subprocess.TimeoutExpired:
|
|
proc.kill()
|
|
stderr = proc.stderr.read() if proc.stderr else ""
|
|
if stderr.strip():
|
|
print("\n--- server stderr ---")
|
|
print(stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_e2e() |