revert: remove MCP Gitea integration, restore direct REST client

Remove all MCP-related infrastructure in favor of direct REST API calls.
MCP added layers without value: Docker container, stdio bridge, hybrid fallback,
healthchecks, SSE transport — all of which added failure modes and token overhead.

Deleted:
- docker/mcp-gitea/docker-compose.yml (MCP container config)
- scripts/mcp-gitea-stdio.cjs (stdio bridge)
- scripts/e2e-mcp-stdio-test*.py (MCP E2E tests)
- scripts/test-kilo-mcp-integration.py
- src/kilocode/agent-manager/mcp-gitea-client.ts (548 lines of MCP wrapper)
- MCP-STDIO-SETUP.md (MCP documentation)
- .vscode/settings.json (hardcoded MCP config with token)
- .kilo/skills/mcp-gitea-connection/ and mcp-gitea.research.md

Restored:
- pipeline-runner.ts: HybridGiteaClient → GiteaClient (direct REST)
  Removed MCP dependency, imports, and initialization.
  No healthcheck waits, no container startup delays.
- process-continuity.md: removed MCP-specific failure modes
- e2e-gns2-test.py: removed Basic Auth, use token auth; fixed spec reference
This commit is contained in:
NW
2026-05-09 01:55:52 +01:00
parent 0f522e61c3
commit 67e8d2e41a
14 changed files with 50 additions and 1697 deletions

View File

@@ -2,23 +2,24 @@
"""GNS-2 End-to-End Integration Test"""
import urllib.request
import json
import base64
import time
import sys
import os
USER, PASS, REPO, ISSUE = 'NW', 'eshkink0t', 'UniqueSoft/APAW', 110
class GiteaAPI:
def __init__(self):
self.base = 'https://git.softuniq.eu/api/v1'
self.creds = base64.b64encode(f"{USER}:{PASS}".encode()).decode()
self.token = os.environ.get('GITEA_TOKEN', '')
def api(self, path, data=None, method='GET'):
url = f"{self.base}/repos/{REPO}{path}"
req = urllib.request.Request(
url, data=json.dumps(data).encode() if data else None,
headers={'Content-Type': 'application/json'}, method=method)
req.add_header('Authorization', f'Basic {self.creds}')
req.add_header('Authorization', f'token {self.token}')
with urllib.request.urlopen(req) as r:
return json.loads(r.read()) if r.status != 204 else None
@@ -130,7 +131,7 @@ def e2e_test():
update_checkpoint('designed', 2, 3500, 4500, 'agent-architect', 'capability-analyst',
' - {agent: arch, invocation: arch-110-001, action: design_spec}')
post_comment('agent-architect', 'subagent_result', 2, 3500, 4500, 'capability-analyst',
"### Spec Designed\n- mcp-gitea-client.ts\n- docker-compose.yml")
"### Spec Designed\n- gitea-client.ts\n- docker-compose.yml")
replace_scoped_label('status', 'status::designed')
add_label('cascade::depth-2')
print("OK")

View File

@@ -1,94 +0,0 @@
#!/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)

View File

@@ -1,105 +0,0 @@
#!/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)

View File

@@ -1,136 +0,0 @@
#!/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)

View File

@@ -1,60 +0,0 @@
#!/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"))

View File

@@ -1,143 +0,0 @@
#!/usr/bin/env python3
"""
test-kilo-mcp-integration.py
Тест интеграции MCP через mcp_settings.json (legacy Kilo Code path).
Проверяет конечную цепочку: mcp_settings.json → stdio bridge → forgejo-mcp → Gitea API
"""
import json
import os
import subprocess
import sys
MCP_SETTINGS_PATH = os.path.expanduser(
"~/.config/Code/User/globalStorage/kilocode.kilo-code/settings/mcp_settings.json"
)
STDIO_SCRIPT = "/home/swp/Projects/APAW/scripts/mcp-gitea-stdio.cjs"
def load_settings():
if not os.path.exists(MCP_SETTINGS_PATH):
raise FileNotFoundError(f"MCP settings not found: {MCP_SETTINGS_PATH}")
with open(MCP_SETTINGS_PATH) as f:
return json.load(f)
def validate_settings(data):
assert "mcpServers" in data, "Missing mcpServers key"
assert "forgejo-gitea" in data["mcpServers"], "Missing forgejo-gitea server"
srv = data["mcpServers"]["forgejo-gitea"]
assert "command" in srv, "Missing command"
assert "args" in srv, "Missing args"
assert "env" in srv, "Missing env"
print(f"✅ Settings valid: command={srv['command']}, args={srv['args']}")
return srv
def test_stdio_rpc(server_config):
print("\n[1] Initialize stdio bridge...")
env = {**os.environ, **server_config.get("env", {})}
cmd = [server_config["command"]] + server_config["args"]
proc = subprocess.Popen(
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
env=env,
cwd="/home/swp/Projects/APAW",
)
def send(method, params=None, call_id=1):
req = json.dumps({"jsonrpc": "2.0", "method": method, "params": params or {}, "id": call_id}) + "\n"
proc.stdin.write(req)
proc.stdin.flush()
def recv():
line = proc.stdout.readline()
return json.loads(line) if line.strip() else None
# initialize
send("initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "test-kilo-mcp", "version": "1.0"}
}, 1)
resp = recv()
assert resp and "result" in resp, f"Initialize failed: {resp}"
print("✅ Initialize OK")
# tools/list
send("tools/list", {}, 2)
resp = recv()
tools = resp.get("result", {}).get("tools", [])
assert len(tools) > 50, f"Expected >50 tools, got {len(tools)}"
print(f"✅ Tools: {len(tools)}")
# get_issue #110
print("\n[2] Call get_issue #110...")
send("tools/call", {
"name": "get_issue",
"arguments": {"owner": "UniqueSoft", "repo": "APAW", "index": 110}
}, 3)
resp = recv()
print(f" Raw resp keys: {list(resp.keys())}")
content_arr = resp.get("result", {}).get("content", [])
print(f" Content array len: {len(content_arr)}")
if content_arr:
content_text = content_arr[0].get("text", "")
print(f" Content text len: {len(content_text)}")
if not content_text:
# fallback: parse result directly
content_text = json.dumps(resp.get("result", {}))
else:
content_text = json.dumps(resp.get("result", {}))
assert content_text, "Empty content"
issue = json.loads(content_text) if content_text.startswith("{") else {"raw": content_text}
if "number" not in issue:
# direct result without wrapper
issue = resp.get("result", {})
assert issue.get("number") == 110, f"Unexpected issue: {issue}"
print(f"✅ Issue #{issue['number']} - {issue.get('title', 'N/A')}")
# checkpoint
print("\n[3] Verify checkpoint in body...")
assert "## GNS Checkpoint" in (issue.get("body") or ""), "No checkpoint"
print("✅ Checkpoint present")
# budget/depth check
print("\n[4] Extract checkpoint YAML...")
body = issue.get("body", "")
import re
match = re.search(r"```yaml\n(.*?)\n```", body, re.S)
assert match, "No YAML block in issue body"
yaml_block = match.group(1)
assert "budget:" in yaml_block, "No budget in checkpoint"
assert "depth:" in yaml_block, "No depth in checkpoint"
print("✅ Budget and depth found in checkpoint")
proc.stdin.close()
proc.wait(timeout=5)
print("✅ Stdio bridge closed cleanly")
def main():
print("=" * 60)
print("Kilo Code MCP Integration Test")
print("=" * 60)
print(f"Settings path: {MCP_SETTINGS_PATH}")
try:
data = load_settings()
srv = validate_settings(data)
test_stdio_rpc(srv)
print("\n" + "=" * 60)
print("✅ ALL KILO MCP INTEGRATION TESTS PASSED")
print("=" * 60)
return 0
except Exception as e:
print(f"\n❌ FAILED: {e}")
import traceback
traceback.print_exc()
return 1
if __name__ == "__main__":
sys.exit(main())