mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-06-15 17:25:26 -04:00
pewdiepie-archdaemon
219 lines
8.1 KiB
Python
Executable File
219 lines
8.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Small Odysseus scoped API helper for Codex terminal sessions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
|
|
def _usage() -> int:
|
|
print("usage:", file=sys.stderr)
|
|
print(" odysseus_api.py capabilities", file=sys.stderr)
|
|
print(" odysseus_api.py todos list", file=sys.stderr)
|
|
print(" odysseus_api.py todos add TITLE", file=sys.stderr)
|
|
print(" odysseus_api.py emails list [limit]", file=sys.stderr)
|
|
print(" odysseus_api.py emails read UID", file=sys.stderr)
|
|
print(" odysseus_api.py emails draft-doc JSON_PAYLOAD", file=sys.stderr)
|
|
print(" odysseus_api.py documents list [limit]", file=sys.stderr)
|
|
print(" odysseus_api.py documents read DOC_ID", file=sys.stderr)
|
|
print(" odysseus_api.py documents create JSON_PAYLOAD", file=sys.stderr)
|
|
print(" odysseus_api.py documents delete DOC_ID", file=sys.stderr)
|
|
print(" odysseus_api.py cookbook tasks", file=sys.stderr)
|
|
print(" odysseus_api.py cookbook servers", file=sys.stderr)
|
|
print(" odysseus_api.py cookbook cached [HOST]", file=sys.stderr)
|
|
print(" odysseus_api.py cookbook presets", file=sys.stderr)
|
|
print(" odysseus_api.py cookbook output SESSION_ID [tail]", file=sys.stderr)
|
|
print(" odysseus_api.py cookbook serve REPO_ID 'CMD' [REMOTE_HOST]", file=sys.stderr)
|
|
print(" odysseus_api.py cookbook preset NAME", file=sys.stderr)
|
|
print(" odysseus_api.py cookbook adopt SESSION_ID MODEL [HOST] [PORT]", file=sys.stderr)
|
|
print(" odysseus_api.py cookbook stop SESSION_ID", file=sys.stderr)
|
|
print(" odysseus_api.py METHOD /api/codex/path [json-body]", file=sys.stderr)
|
|
return 2
|
|
|
|
|
|
def _config() -> tuple[str, str] | None:
|
|
base_url = os.environ.get("ODYSSEUS_URL", "").strip().rstrip("/")
|
|
token = os.environ.get("ODYSSEUS_API_TOKEN", "").strip()
|
|
missing = []
|
|
if not base_url:
|
|
missing.append("ODYSSEUS_URL")
|
|
if not token:
|
|
missing.append("ODYSSEUS_API_TOKEN")
|
|
if missing:
|
|
print(f"missing {', '.join(missing)}; create a Codex Agent token in Odysseus Settings", file=sys.stderr)
|
|
return None
|
|
return base_url, token
|
|
|
|
|
|
def main() -> int:
|
|
if len(sys.argv) < 2:
|
|
return _usage()
|
|
|
|
command = sys.argv[1].lower()
|
|
if command == "capabilities":
|
|
method = "GET"
|
|
path = "/api/codex/capabilities"
|
|
body = None
|
|
elif command == "todos":
|
|
if len(sys.argv) < 3:
|
|
return _usage()
|
|
action = sys.argv[2].lower()
|
|
path = "/api/codex/todos"
|
|
if action == "list":
|
|
method = "GET"
|
|
body = None
|
|
elif action == "add" and len(sys.argv) >= 4:
|
|
method = "POST"
|
|
body = json.dumps({"action": "add", "title": " ".join(sys.argv[3:])})
|
|
else:
|
|
return _usage()
|
|
elif command == "emails":
|
|
if len(sys.argv) < 3:
|
|
return _usage()
|
|
action = sys.argv[2].lower()
|
|
if action == "list":
|
|
method = "GET"
|
|
limit = sys.argv[3] if len(sys.argv) >= 4 else "10"
|
|
path = f"/api/codex/emails?folder=INBOX&limit={limit}&offset=0&filter=all"
|
|
body = None
|
|
elif action == "read" and len(sys.argv) >= 4:
|
|
method = "GET"
|
|
path = f"/api/codex/emails/{sys.argv[3]}"
|
|
body = None
|
|
elif action in ("draft-doc", "draft_document") and len(sys.argv) >= 4:
|
|
method = "POST"
|
|
path = "/api/codex/emails/draft-document"
|
|
body = " ".join(sys.argv[3:])
|
|
else:
|
|
return _usage()
|
|
elif command in ("documents", "docs"):
|
|
if len(sys.argv) < 3:
|
|
return _usage()
|
|
action = sys.argv[2].lower()
|
|
if action == "list":
|
|
method = "GET"
|
|
limit = sys.argv[3] if len(sys.argv) >= 4 else "50"
|
|
path = f"/api/codex/documents?limit={limit}"
|
|
body = None
|
|
elif action == "read" and len(sys.argv) >= 4:
|
|
method = "GET"
|
|
path = f"/api/codex/documents/{sys.argv[3]}"
|
|
body = None
|
|
elif action == "create" and len(sys.argv) >= 4:
|
|
method = "POST"
|
|
path = "/api/codex/documents"
|
|
body = " ".join(sys.argv[3:])
|
|
elif action == "delete" and len(sys.argv) >= 4:
|
|
method = "DELETE"
|
|
path = f"/api/codex/documents/{sys.argv[3]}"
|
|
body = None
|
|
else:
|
|
return _usage()
|
|
elif command == "cookbook":
|
|
if len(sys.argv) < 3:
|
|
return _usage()
|
|
action = sys.argv[2].lower()
|
|
if action == "tasks":
|
|
method = "GET"
|
|
path = "/api/codex/cookbook/tasks"
|
|
body = None
|
|
elif action == "servers":
|
|
method = "GET"
|
|
path = "/api/codex/cookbook/servers"
|
|
body = None
|
|
elif action == "output" and len(sys.argv) >= 4:
|
|
method = "GET"
|
|
sid = sys.argv[3]
|
|
tail = sys.argv[4] if len(sys.argv) >= 5 else "400"
|
|
path = f"/api/codex/cookbook/output/{sid}?tail={tail}"
|
|
body = None
|
|
elif action == "cached":
|
|
method = "GET"
|
|
if len(sys.argv) >= 4:
|
|
from urllib.parse import quote
|
|
path = f"/api/codex/cookbook/cached?host={quote(sys.argv[3])}"
|
|
else:
|
|
path = "/api/codex/cookbook/cached"
|
|
body = None
|
|
elif action == "presets":
|
|
method = "GET"
|
|
path = "/api/codex/cookbook/presets"
|
|
body = None
|
|
elif action == "preset" and len(sys.argv) >= 4:
|
|
from urllib.parse import quote
|
|
method = "POST"
|
|
path = f"/api/codex/cookbook/preset/{quote(sys.argv[3])}"
|
|
body = None
|
|
elif action == "adopt" and len(sys.argv) >= 5:
|
|
method = "POST"
|
|
path = "/api/codex/cookbook/adopt"
|
|
payload = {"tmux_session": sys.argv[3], "model": sys.argv[4]}
|
|
if len(sys.argv) >= 6: payload["host"] = sys.argv[5]
|
|
if len(sys.argv) >= 7: payload["port"] = int(sys.argv[6])
|
|
body = json.dumps(payload)
|
|
elif action == "serve" and len(sys.argv) >= 5:
|
|
method = "POST"
|
|
path = "/api/codex/cookbook/serve"
|
|
payload = {"repo_id": sys.argv[3], "cmd": sys.argv[4]}
|
|
if len(sys.argv) >= 6:
|
|
payload["remote_host"] = sys.argv[5]
|
|
body = json.dumps(payload)
|
|
elif action == "stop" and len(sys.argv) >= 4:
|
|
method = "POST"
|
|
path = f"/api/codex/cookbook/stop/{sys.argv[3]}"
|
|
body = None
|
|
else:
|
|
return _usage()
|
|
else:
|
|
if len(sys.argv) < 3:
|
|
return _usage()
|
|
method = sys.argv[1].upper()
|
|
path = sys.argv[2]
|
|
body = sys.argv[3] if len(sys.argv) > 3 else None
|
|
|
|
if not path.startswith("/"):
|
|
path = "/" + path
|
|
if not path.startswith("/api/codex/"):
|
|
print("refusing non-/api/codex path; use scoped Odysseus integration endpoints only", file=sys.stderr)
|
|
return 2
|
|
|
|
config = _config()
|
|
if config is None:
|
|
return 2
|
|
base_url, token = config
|
|
|
|
data = None
|
|
headers = {
|
|
"Accept": "application/json",
|
|
"Authorization": f"Bearer {token}",
|
|
}
|
|
if body is not None:
|
|
try:
|
|
parsed = json.loads(body)
|
|
except json.JSONDecodeError as exc:
|
|
print(f"invalid json body: {exc}", file=sys.stderr)
|
|
return 2
|
|
data = json.dumps(parsed).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
|
|
req = urllib.request.Request(base_url + path, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=20) as resp:
|
|
print(resp.read().decode("utf-8"))
|
|
return 0
|
|
except urllib.error.HTTPError as exc:
|
|
text = exc.read().decode("utf-8", errors="replace")
|
|
print(text or f"HTTP {exc.code}", file=sys.stderr)
|
|
return 1
|
|
except OSError as exc:
|
|
print(f"request failed: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|