93 lines
2.4 KiB
Python
93 lines
2.4 KiB
Python
|
import base64
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
import os
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
|
||
|
|
log = logging.getLogger("forgemcp")
|
||
|
|
|
||
|
|
DEFAULT_BASE_URL = "https://forge.mql5.io"
|
||
|
|
DEFAULT_TIMEOUT = 30
|
||
|
|
|
||
|
|
_CONFIG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json")
|
||
|
|
|
||
|
|
_client = None
|
||
|
|
|
||
|
|
|
||
|
|
def load_config():
|
||
|
|
if not os.path.exists(_CONFIG_PATH):
|
||
|
|
return {}
|
||
|
|
with open(_CONFIG_PATH, "r") as f:
|
||
|
|
return json.load(f)
|
||
|
|
|
||
|
|
|
||
|
|
def get_client():
|
||
|
|
"""Return a singleton authenticated httpx client for the Algo Forge API."""
|
||
|
|
global _client
|
||
|
|
if _client is not None:
|
||
|
|
return _client
|
||
|
|
|
||
|
|
config = load_config()
|
||
|
|
base_url = config.get("base_url") or DEFAULT_BASE_URL
|
||
|
|
token = config.get("token") or os.environ.get("FORGE_TOKEN")
|
||
|
|
timeout = config.get("timeout") or DEFAULT_TIMEOUT
|
||
|
|
|
||
|
|
if not token:
|
||
|
|
log.warning("No Algo Forge token configured (config.json or FORGE_TOKEN).")
|
||
|
|
|
||
|
|
headers = {"Accept": "application/json"}
|
||
|
|
if token:
|
||
|
|
headers["Authorization"] = f"token {token}"
|
||
|
|
|
||
|
|
_client = httpx.Client(
|
||
|
|
base_url=f"{base_url.rstrip('/')}/api/v1",
|
||
|
|
headers=headers,
|
||
|
|
timeout=timeout,
|
||
|
|
)
|
||
|
|
return _client
|
||
|
|
|
||
|
|
|
||
|
|
def request(method, path, **kwargs):
|
||
|
|
"""Make an API call and normalize the result into a dict or list.
|
||
|
|
|
||
|
|
Returns parsed JSON on success, or {"error": "..."} on any failure.
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
resp = get_client().request(method, path, **kwargs)
|
||
|
|
except httpx.RequestError as exc:
|
||
|
|
return {"error": f"Connection to Algo Forge failed: {exc}"}
|
||
|
|
|
||
|
|
if resp.status_code == 204:
|
||
|
|
return {"ok": True}
|
||
|
|
|
||
|
|
if resp.status_code >= 400:
|
||
|
|
return {"error": _error_message(resp)}
|
||
|
|
|
||
|
|
if not resp.content:
|
||
|
|
return {"ok": True}
|
||
|
|
|
||
|
|
try:
|
||
|
|
return resp.json()
|
||
|
|
except ValueError:
|
||
|
|
return {"error": "Algo Forge returned a non-JSON response."}
|
||
|
|
|
||
|
|
|
||
|
|
def _error_message(resp):
|
||
|
|
"""Extract a human-readable error from a Forgejo error response."""
|
||
|
|
try:
|
||
|
|
body = resp.json()
|
||
|
|
msg = body.get("message") or body.get("error") or str(body)
|
||
|
|
except ValueError:
|
||
|
|
msg = resp.text or "unknown error"
|
||
|
|
return f"HTTP {resp.status_code}: {msg}"
|
||
|
|
|
||
|
|
|
||
|
|
def encode_content(text):
|
||
|
|
"""Base64-encode file text for the contents API."""
|
||
|
|
return base64.b64encode(text.encode("utf-8")).decode("ascii")
|
||
|
|
|
||
|
|
|
||
|
|
def decode_content(b64):
|
||
|
|
"""Decode base64 file content returned by the contents API."""
|
||
|
|
return base64.b64decode(b64).decode("utf-8", errors="replace")
|