mirror of
https://github.com/allaunthefox/SilverSight.git
synced 2026-07-31 01:25:21 +00:00
allaun
85141a4b94
- Port NUVMAP projection engine from Research Stack to SilverSight with Q16_16 fixed-point (zero Float) and CBOR serialization - Add Rotational Wave — Braid Correspondence formalization at boundary (ChiralLabel, RossbyDrift, rossby_convergence_bound stubbed, kelvin_wave_eigensolid proven) - Add auto-pipeline CI workflow, webhook receiver, Forgejo MCP server - Add SOPS/Age encryption config - Add stack compose for portable deployment - Add rotational wave design doc
88 lines
4.5 KiB
Python
88 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Forgejo MCP Server — AI agent interface to Forgejo API."""
|
|
import json, sys, os, urllib.request, urllib.error
|
|
|
|
BASE = os.environ.get("FORGEJO_URL", "http://localhost:3000")
|
|
TOKEN = os.environ.get("FORGEJO_TOKEN", "858fe52d74764f20f23743a8d04544b95212985d")
|
|
|
|
def api(method, path, data=None):
|
|
req = urllib.request.Request(
|
|
f"{BASE}{path}",
|
|
data=json.dumps(data).encode() if data else None,
|
|
headers={
|
|
"Authorization": f"token {TOKEN}",
|
|
"Content-Type": "application/json"
|
|
},
|
|
method=method
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req) as r:
|
|
return json.loads(r.read())
|
|
except urllib.error.HTTPError as e:
|
|
return {"error": e.code, "body": e.read().decode()[:200]}
|
|
|
|
def handle_request(req):
|
|
method = req.get("method", "")
|
|
params = req.get("params", {})
|
|
|
|
if method == "list_tools":
|
|
return {"tools": [
|
|
{"name": "forgejo_search_repos", "description": "Search repositories", "inputSchema": {"type": "object", "properties": {"query": {"type": "string"}}}},
|
|
{"name": "forgejo_get_repo", "description": "Get repository details", "inputSchema": {"type": "object", "properties": {"owner": {"type": "string"}, "repo": {"type": "string"}}}},
|
|
{"name": "forgejo_list_issues", "description": "List repository issues", "inputSchema": {"type": "object", "properties": {"owner": {"type": "string"}, "repo": {"type": "string"}, "state": {"type": "string"}}}},
|
|
{"name": "forgejo_create_issue", "description": "Create an issue", "inputSchema": {"type": "object", "properties": {"owner": {"type": "string"}, "repo": {"type": "string"}, "title": {"type": "string"}, "body": {"type": "string"}}}},
|
|
{"name": "forgejo_list_pull_requests", "description": "List pull requests", "inputSchema": {"type": "object", "properties": {"owner": {"type": "string"}, "repo": {"type": "string"}, "state": {"type": "string"}}}},
|
|
{"name": "forgejo_get_file", "description": "Get file content from repo", "inputSchema": {"type": "object", "properties": {"owner": {"type": "string"}, "repo": {"type": "string"}, "filepath": {"type": "string"}, "ref": {"type": "string"}}}},
|
|
{"name": "forgejo_push_mirror", "description": "Push mirror to Forgejo", "inputSchema": {"type": "object", "properties": {"owner": {"type": "string"}, "repo": {"type": "string"}, "remote_url": {"type": "string"}}}},
|
|
]}
|
|
|
|
elif method == "call_tool":
|
|
name = params.get("name", "")
|
|
args = params.get("arguments", {})
|
|
|
|
if name == "forgejo_search_repos":
|
|
result = api("GET", f"/api/v1/repos/search?q={args.get('query','')}")
|
|
elif name == "forgejo_get_repo":
|
|
result = api("GET", f"/api/v1/repos/{args['owner']}/{args['repo']}")
|
|
elif name == "forgejo_list_issues":
|
|
result = api("GET", f"/api/v1/repos/{args['owner']}/{args['repo']}/issues?state={args.get('state','open')}")
|
|
elif name == "forgejo_create_issue":
|
|
result = api("POST", f"/api/v1/repos/{args['owner']}/{args['repo']}/issues", {"title": args["title"], "body": args.get("body","")})
|
|
elif name == "forgejo_list_pull_requests":
|
|
result = api("GET", f"/api/v1/repos/{args['owner']}/{args['repo']}/pulls?state={args.get('state','open')}")
|
|
elif name == "forgejo_get_file":
|
|
result = api("GET", f"/api/v1/repos/{args['owner']}/{args['repo']}/contents/{args['filepath']}?ref={args.get('ref','main')}")
|
|
elif name == "forgejo_push_mirror":
|
|
result = api("POST", f"/api/v1/repos/migrate", {
|
|
"clone_addr": args["remote_url"],
|
|
"repo_name": args["repo"],
|
|
"uid": 1,
|
|
"mirror": True,
|
|
"private": True
|
|
})
|
|
else:
|
|
result = {"error": f"unknown tool: {name}"}
|
|
|
|
return {"content": [{"type": "text", "text": json.dumps(result, indent=2)}]}
|
|
|
|
elif method == "list_resource_templates":
|
|
return {"resourceTemplates": []}
|
|
|
|
return {"error": f"unknown method: {method}"}
|
|
|
|
def main():
|
|
while True:
|
|
line = sys.stdin.readline()
|
|
if not line:
|
|
break
|
|
try:
|
|
req = json.loads(line)
|
|
resp = handle_request(req)
|
|
sys.stdout.write(json.dumps(resp) + "\n")
|
|
sys.stdout.flush()
|
|
except Exception as e:
|
|
sys.stdout.write(json.dumps({"error": str(e)}) + "\n")
|
|
sys.stdout.flush()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|