feat(gns-2): stdio MCP transport with hybrid fallback
This commit is contained in:
NW
94
scripts/e2e-mcp-stdio-test-v2.py
Normal file
94
scripts/e2e-mcp-stdio-test-v2.py
Normal file
@@ -0,0 +1,94 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
e2e-mcp-stdio-test-v2.py
|
||||
Minimal E2E test for MCP stdio transport via @ric_/forgejo-mcp.
|
||||
Uses subprocess.communicate() to avoid pipe deadlock.
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
import json
|
||||
import sys
|
||||
import base64
|
||||
|
||||
STDIO_CMD = ["bunx", "@ric_/forgejo-mcp"]
|
||||
GITEA_API = "https://git.softuniq.eu/api/v1/repos/UniqueSoft/APAW"
|
||||
USER, PASS = "NW", "eshkink0t"
|
||||
|
||||
def test_stdio():
|
||||
print("="*60)
|
||||
print("E2E MCP Stdio Test v2")
|
||||
print("="*60)
|
||||
|
||||
env = {
|
||||
**subprocess.os.environ,
|
||||
"FORGEJO_URL": "https://git.softuniq.eu",
|
||||
"FORGEJO_TOKEN": PASS,
|
||||
"LOG_LEVEL": "warn",
|
||||
}
|
||||
|
||||
# 1. Initialize
|
||||
print("\n[1] Initialize...")
|
||||
proc = subprocess.Popen(
|
||||
STDIO_CMD,
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
env=env,
|
||||
)
|
||||
req = json.dumps({"jsonrpc": "2.0", "method": "initialize", "params": {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "e2e-test", "version": "1.0"}}, "id": 1})
|
||||
out, err = proc.communicate(input=req + "\n")
|
||||
print("stderr:", err.strip()[:200])
|
||||
resp = json.loads(out.strip().splitlines()[-1])
|
||||
assert resp["result"]["serverInfo"]["name"] == "forgejo-mcp", f"Unexpected: {resp}"
|
||||
print("✅ Initialize OK")
|
||||
|
||||
# 2. tools/list
|
||||
print("\n[2] List tools...")
|
||||
proc2 = subprocess.Popen(STDIO_CMD, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env)
|
||||
out2, err2 = proc2.communicate(
|
||||
input=json.dumps({"jsonrpc":"2.0","method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"t","version":"1"}},"id":1}) + "\n" +
|
||||
json.dumps({"jsonrpc":"2.0","method":"tools/list","params":{},"id":2}) + "\n"
|
||||
)
|
||||
lines = [l for l in out2.strip().splitlines() if l.strip()]
|
||||
resp2 = json.loads(lines[-1])
|
||||
tools = resp2.get("result", {}).get("tools", [])
|
||||
assert len(tools) > 50, f"Expected >50 tools, got {len(tools)}"
|
||||
print(f"✅ Tools: {len(tools)}")
|
||||
|
||||
# 3. get_issue
|
||||
print("\n[3] gitea_get_issue #110...")
|
||||
proc3 = subprocess.Popen(STDIO_CMD, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env)
|
||||
out3, err3 = proc3.communicate(
|
||||
input=json.dumps({"jsonrpc":"2.0","method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"t","version":"1"}},"id":1}) + "\n" +
|
||||
json.dumps({"jsonrpc":"2.0","method":"tools/call","params":{"name":"get_issue","arguments":{"owner":"UniqueSoft","repo":"APAW","issue_number":110}},"id":3}) + "\n"
|
||||
)
|
||||
lines3 = [l for l in out3.strip().splitlines() if l.strip()]
|
||||
resp3 = json.loads(lines3[-1])
|
||||
content = json.loads(resp3["result"]["content"][0]["text"])
|
||||
assert content.get("number") == 110, f"Unexpected issue: {content}"
|
||||
print(f"✅ Issue #{content['number']} - {content.get('title','N/A')}")
|
||||
|
||||
# 4. REST consistency
|
||||
print("\n[4] REST consistency...")
|
||||
import urllib.request
|
||||
creds = base64.b64encode(f"{USER}:{PASS}".encode()).decode()
|
||||
req4 = urllib.request.Request(f"{GITEA_API}/issues/110", headers={"Accept": "application/json", "Authorization": f"Basic {creds}"})
|
||||
with urllib.request.urlopen(req4) as r:
|
||||
rest = json.loads(r.read())
|
||||
assert rest["title"] == content["title"], "Title mismatch"
|
||||
print("✅ REST consistent")
|
||||
|
||||
print("\n" + "="*60)
|
||||
print("✅ ALL E2E MCP STDIO TESTS PASSED")
|
||||
print("="*60)
|
||||
return 0
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.exit(test_stdio())
|
||||
except Exception as e:
|
||||
print(f"\n❌ FAILED: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
105
scripts/e2e-mcp-stdio-test-v3.py
Normal file
105
scripts/e2e-mcp-stdio-test-v3.py
Normal file
@@ -0,0 +1,105 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
e2e-mcp-stdio-test-v3.py
|
||||
E2E test with correct tool names from forgejo-mcp.
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
import json
|
||||
import sys
|
||||
import base64
|
||||
|
||||
STDIO_CMD = ["bunx", "@ric_/forgejo-mcp"]
|
||||
GITEA_API = "https://git.softuniq.eu/api/v1/repos/UniqueSoft/APAW"
|
||||
USER, PASS = "NW", "eshkink0t"
|
||||
|
||||
def call_stdio(method, params=None, call_id=1):
|
||||
env = {
|
||||
**subprocess.os.environ,
|
||||
"FORGEJO_URL": "https://git.softuniq.eu",
|
||||
"FORGEJO_TOKEN": "ad1176845d1170f840193a700eb5319998c52601", # Personal access token instead of password
|
||||
"LOG_LEVEL": "warn",
|
||||
}
|
||||
msgs = [
|
||||
json.dumps({"jsonrpc":"2.0","method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"t","version":"1"}},"id":1}),
|
||||
]
|
||||
if method == "tools/list":
|
||||
msgs.append(json.dumps({"jsonrpc":"2.0","method":"tools/list","params":{},"id":call_id}))
|
||||
elif method == "tools/call":
|
||||
msgs.append(json.dumps({"jsonrpc":"2.0","method":"tools/call","params":params,"id":call_id}))
|
||||
proc = subprocess.Popen(STDIO_CMD, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env)
|
||||
out, err = proc.communicate(input="\n".join(msgs) + "\n")
|
||||
lines = [l for l in out.strip().splitlines() if l.strip()]
|
||||
return json.loads(lines[-1]) if lines else None, err
|
||||
|
||||
def test_stdio():
|
||||
print("="*60)
|
||||
print("E2E MCP Stdio Test v3")
|
||||
print("="*60)
|
||||
|
||||
# 1. Initialize
|
||||
print("\n[1] Initialize...")
|
||||
resp, err = call_stdio("initialize")
|
||||
assert resp["result"]["serverInfo"]["name"] == "forgejo-mcp"
|
||||
print("✅ Initialize OK")
|
||||
|
||||
# 2. tools/list
|
||||
print("\n[2] List tools...")
|
||||
resp2, err2 = call_stdio("tools/list", call_id=2)
|
||||
tools = resp2.get("result", {}).get("tools", [])
|
||||
assert len(tools) > 50, f"Got {len(tools)}"
|
||||
tool_names = [t["name"] for t in tools]
|
||||
print(f"✅ Tools: {len(tools)}")
|
||||
issue_tool = None
|
||||
for t in tool_names:
|
||||
if "issue" in t and "list" not in t and "comment" not in t and "label" not in t:
|
||||
issue_tool = t
|
||||
break
|
||||
print(f" Issue tool candidate: {issue_tool}")
|
||||
|
||||
# 3. get_issue
|
||||
print("\n[3] Fetch issue #110...")
|
||||
for tool_name in ["get_issue", "gitea_get_issue"]:
|
||||
resp3, err3 = call_stdio("tools/call", params={"name": tool_name, "arguments": {"owner": "UniqueSoft", "repo": "APAW", "index": 110}}, call_id=3)
|
||||
content_text = resp3.get("result", {}).get("content", [{}])[0].get("text", "")
|
||||
if content_text and content_text.strip():
|
||||
print(f" Tool '{tool_name}' returned data")
|
||||
print(f" Content text length: {len(content_text)}")
|
||||
print(f" First 500 chars of content: {repr(content_text[:500])}")
|
||||
break
|
||||
else:
|
||||
print(f" Tool responses: {resp3}")
|
||||
raise Exception("No tool returned data")
|
||||
|
||||
issue_data = json.loads(content_text)
|
||||
assert issue_data.get("number") == 110, f"Unexpected: {issue_data}"
|
||||
print(f"✅ Issue #{issue_data['number']} - {issue_data.get('title','N/A')}")
|
||||
|
||||
# 4. Verify checkpoint
|
||||
print("\n[4] Verify checkpoint...")
|
||||
assert "## GNS Checkpoint" in (issue_data.get("body") or ""), "No checkpoint"
|
||||
print("✅ Checkpoint present")
|
||||
|
||||
# 5. REST consistency
|
||||
print("\n[5] REST consistency...")
|
||||
import urllib.request
|
||||
creds = base64.b64encode(f"{USER}:{PASS}".encode()).decode()
|
||||
req = urllib.request.Request(f"{GITEA_API}/issues/110", headers={"Accept": "application/json", "Authorization": f"Basic {creds}"})
|
||||
with urllib.request.urlopen(req) as r:
|
||||
rest = json.loads(r.read())
|
||||
assert rest["title"] == issue_data["title"], "Mismatch"
|
||||
print("✅ REST consistent")
|
||||
|
||||
print("\n" + "="*60)
|
||||
print("✅ ALL E2E MCP STDIO TESTS PASSED")
|
||||
print("="*60)
|
||||
return 0
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.exit(test_stdio())
|
||||
except Exception as e:
|
||||
print(f"\n❌ FAILED: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
136
scripts/e2e-mcp-stdio-test.py
Executable file
136
scripts/e2e-mcp-stdio-test.py
Executable file
@@ -0,0 +1,136 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
e2e-mcp-stdio-test.py
|
||||
End-to-end test for MCP Gitea stdio transport.
|
||||
|
||||
1. Spawn stdio bridge via bun
|
||||
2. Call initialize
|
||||
3. Call tools/list
|
||||
4. Call tools/call gitea_get_issue for issue #110
|
||||
5. Validate response
|
||||
6. Compare with REST API fallback
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
import json
|
||||
import sys
|
||||
import time
|
||||
|
||||
STDIO_CMD = ["bun", "scripts/mcp-gitea-stdio.cjs"]
|
||||
GITEA_API = "https://git.softuniq.eu/api/v1/repos/UniqueSoft/APAW"
|
||||
USER, PASS = "NW", "eshkink0t"
|
||||
|
||||
def main():
|
||||
print("="*60)
|
||||
print("E2E MCP Stdio Test")
|
||||
print("="*60)
|
||||
|
||||
proc = subprocess.Popen(
|
||||
STDIO_CMD,
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
cwd="/home/swp/Projects/APAW",
|
||||
)
|
||||
|
||||
def send(msg):
|
||||
line = json.dumps(msg) + "\n"
|
||||
proc.stdin.write(line)
|
||||
proc.stdin.flush()
|
||||
print(f"→ {line.strip()}")
|
||||
|
||||
def recv():
|
||||
line = proc.stdout.readline()
|
||||
print(f"← {line.strip()}")
|
||||
return json.loads(line)
|
||||
|
||||
# 1. Initialize
|
||||
print("\n[1] Initialize...")
|
||||
send({
|
||||
"jsonrpc": "2.0",
|
||||
"method": "initialize",
|
||||
"params": {
|
||||
"protocolVersion": "2024-05-08",
|
||||
"capabilities": {},
|
||||
"clientInfo": {"name": "e2e-test-client", "version": "1.0.0"}
|
||||
},
|
||||
"id": 1
|
||||
})
|
||||
resp = recv()
|
||||
assert resp["result"]["serverInfo"]["name"] == "forgejo-mcp", "Unexpected server name"
|
||||
print("✅ Initialize OK")
|
||||
|
||||
# 2. tools/list
|
||||
print("\n[2] List tools...")
|
||||
send({"jsonrpc": "2.0", "method": "tools/list", "params": {}, "id": 2})
|
||||
resp = recv()
|
||||
tools = resp.get("result", {}).get("tools", [])
|
||||
assert len(tools) > 50, f"Expected >50 tools, got {len(tools)}"
|
||||
print(f"✅ Tools listed: {len(tools)}")
|
||||
|
||||
# 3. tools/call get_issue
|
||||
print("\n[3] Call get_issue #110...")
|
||||
send({
|
||||
"jsonrpc": "2.0",
|
||||
"method": "tools/call",
|
||||
"params": {
|
||||
"name": "get_issue",
|
||||
"arguments": {
|
||||
"owner": "UniqueSoft",
|
||||
"repo": "APAW",
|
||||
"index": 110
|
||||
}
|
||||
},
|
||||
"id": 3
|
||||
})
|
||||
resp = recv()
|
||||
print(f"DEBUG: Response received: {resp}")
|
||||
result_content = resp["result"]["content"]
|
||||
print(f"DEBUG: Result content: {result_content}")
|
||||
result_text = result_content[0]["text"]
|
||||
print(f"DEBUG: Result text: {result_text}")
|
||||
issue_data = json.loads(result_text)
|
||||
assert issue_data["number"] == 110, f"Expected issue 110, got {issue_data.get('number')}"
|
||||
print(f"✅ Issue fetched: #{issue_data['number']} - {issue_data.get('title', 'N/A')}")
|
||||
|
||||
# 4. Verify checkpoint exists in issue body
|
||||
print("\n[4] Verify checkpoint in issue body...")
|
||||
assert "## GNS Checkpoint" in (issue_data.get("body") or ""), "Checkpoint not found in issue body"
|
||||
print("✅ Checkpoint found")
|
||||
|
||||
# 5. Compare with REST API for consistency
|
||||
print("\n[5] REST API consistency check...")
|
||||
import urllib.request
|
||||
import base64
|
||||
creds = base64.b64encode(f"{USER}:{PASS}".encode()).decode()
|
||||
req = urllib.request.Request(
|
||||
f"{GITEA_API}/issues/110",
|
||||
headers={"Accept": "application/json", "Authorization": f"Basic {creds}"}
|
||||
)
|
||||
with urllib.request.urlopen(req) as r:
|
||||
rest_issue = json.loads(r.read())
|
||||
assert rest_issue["number"] == issue_data["number"], "MCP and REST issue numbers differ"
|
||||
assert rest_issue["title"] == issue_data["title"], "MCP and REST issue titles differ"
|
||||
print("✅ REST API consistent")
|
||||
|
||||
# 6. Close gracefully
|
||||
print("\n[6] Terminate stdio bridge...")
|
||||
proc.stdin.close()
|
||||
proc.wait(timeout=5)
|
||||
print("✅ Stdio bridge closed")
|
||||
|
||||
print("\n" + "="*60)
|
||||
print("✅ ALL E2E MCP STDIO TESTS PASSED")
|
||||
print("="*60)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.exit(main())
|
||||
except Exception as e:
|
||||
print(f"\n❌ FAILED: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
60
scripts/mcp-gitea-stdio.cjs
Normal file
60
scripts/mcp-gitea-stdio.cjs
Normal file
@@ -0,0 +1,60 @@
|
||||
#!/usr/bin/env bun
|
||||
/**
|
||||
* mcp-gitea-stdio.cjs
|
||||
* MCP Stdio Bridge — wraps @ric_/forgejo-mcp for Kilo Code infrastructure
|
||||
*
|
||||
* This replaces HTTP↔SSE fallback complexity with direct stdio invocation
|
||||
* of the official forgejo-mcp package.
|
||||
*
|
||||
* Usage: MCP_STDIO_COMMAND="bun scripts/mcp-gitea-stdio.cjs"
|
||||
* Or: FORGEJO_TOKEN=xxx bun scripts/mcp-gitea-stdio.cjs
|
||||
*/
|
||||
|
||||
import { spawn } from "child_process"
|
||||
|
||||
const FORGEJO_TOKEN = process.env.FORGEJO_TOKEN || process.env.GITEA_TOKEN || ""
|
||||
const FORGEJO_URL = process.env.FORGEJO_URL || "https://git.softuniq.eu"
|
||||
const USE_CONTAINER = process.env.USE_MCP_CONTAINER === "1"
|
||||
|
||||
let child = null
|
||||
|
||||
function log(...args) {
|
||||
// eslint-disable-next-line no-console
|
||||
console.error("[stdio]", ...args)
|
||||
}
|
||||
|
||||
log("Starting forgejo-mcp stdio bridge...")
|
||||
|
||||
if (!FORGEJO_TOKEN) {
|
||||
log("WARNING: FORGEJO_TOKEN not set. MCP tools will fail authentication.")
|
||||
}
|
||||
|
||||
if (USE_CONTAINER) {
|
||||
// Spawn Docker container with stdio passthrough
|
||||
child = spawn(
|
||||
"docker", ["exec", "-i", "mcp-gitea", "node", "dist/index.js"],
|
||||
{ env: { ...process.env, FORGEJO_TOKEN, FORGEJO_URL } }
|
||||
)
|
||||
} else {
|
||||
child = spawn(
|
||||
"bunx", ["@ric_/forgejo-mcp"],
|
||||
{ env: { ...process.env, FORGEJO_URL, FORGEJO_TOKEN, LOG_LEVEL: "warn" } }
|
||||
)
|
||||
}
|
||||
|
||||
process.stdin.pipe(child.stdin)
|
||||
child.stdout.pipe(process.stdout)
|
||||
child.stderr.pipe(process.stderr)
|
||||
|
||||
child.on("exit", (code) => {
|
||||
log("forgejo-mcp exited with code", code)
|
||||
process.exit(code || 0)
|
||||
})
|
||||
|
||||
child.on("error", (err) => {
|
||||
log("Failed to start forgejo-mcp:", err.message)
|
||||
process.exit(1)
|
||||
})
|
||||
|
||||
process.on("SIGTERM", () => child && child.kill("SIGTERM"))
|
||||
process.on("SIGINT", () => child && child.kill("SIGINT"))
|
||||
Reference in New Issue
Block a user