freshservice-claude/scripts/fs.py

#!/usr/bin/env python3
"""Freshservice API helper for the freshservice skill.

Handles auth, base URL, query encoding, JSON bodies, and pagination so the
model only has to choose an HTTP method, a path, and parameters.

Credentials (checked in this order):
  1. Environment: FRESHSERVICE_DOMAIN, FRESHSERVICE_API_KEY
  2. File ~/.freshservice/credentials with lines:
        FRESHSERVICE_DOMAIN=acme
        FRESHSERVICE_API_KEY=xxxxxxxx
  FRESHSERVICE_DOMAIN may be the subdomain ("acme") or full host.

Usage:
  fs.py check
  fs.py request GET /tickets --param per_page=5
  fs.py request GET /tickets/filter --param 'query=status:2 AND priority:3'
  fs.py request GET /tickets/123 --param include=conversations
  fs.py request POST /tickets --data '{"subject":"...","description":"...","email":"a@b.com","priority":2,"status":2}'
  fs.py request PUT /tickets/123 --data '{"status":4}'
  fs.py request POST /tickets/123/reply --data '{"body":"Hi"}'

Notes:
  * "/api/v2" is prepended to PATH automatically if absent.
  * For the filter endpoints the API requires the value wrapped in double
    quotes (?query="..."). This helper auto-wraps the `query` and `filter`
    params, so pass them unquoted.
"""

from __future__ import annotations

import argparse
import base64
import json
import os
import re
import sys
import urllib.error
import urllib.parse
import urllib.request


def load_credentials() -> tuple[str, str]:
    domain = os.environ.get("FRESHSERVICE_DOMAIN", "").strip()
    api_key = os.environ.get("FRESHSERVICE_API_KEY", "").strip()
    cred_file = os.path.expanduser("~/.freshservice/credentials")
    if (not domain or not api_key) and os.path.exists(cred_file):
        with open(cred_file) as fh:
            for line in fh:
                line = line.strip()
                if not line or line.startswith("#") or "=" not in line:
                    continue
                k, v = line.split("=", 1)
                k, v = k.strip(), v.strip().strip('"').strip("'")
                if k == "FRESHSERVICE_DOMAIN" and not domain:
                    domain = v
                elif k == "FRESHSERVICE_API_KEY" and not api_key:
                    api_key = v
    if not domain or not api_key:
        sys.exit(
            "ERROR: Missing credentials. Set FRESHSERVICE_DOMAIN and "
            "FRESHSERVICE_API_KEY in the environment or in "
            "~/.freshservice/credentials (see the skill's SETUP section)."
        )
    if "." not in domain:
        domain = f"{domain}.freshservice.com"
    return domain, api_key


def request(method: str, path: str, params: dict[str, str], data: str | None) -> dict:
    domain, api_key = load_credentials()
    path = path if path.startswith("/") else f"/{path}"
    if not path.startswith("/api/"):
        path = f"/api/v2{path}"
    url = f"https://{domain}{path}"

    # The filter endpoints require the value double-quoted in the URL.
    encoded = {}
    for k, v in params.items():
        if k in ("query", "filter") and not (v.startswith('"') and v.endswith('"')):
            v = f'"{v}"'
        encoded[k] = v
    if encoded:
        url += "?" + urllib.parse.urlencode(encoded)

    token = base64.b64encode(f"{api_key}:X".encode()).decode()
    headers = {
        "Authorization": f"Basic {token}",
        "Content-Type": "application/json",
        "Accept": "application/json",
    }
    body = data.encode() if data else None
    req = urllib.request.Request(url, data=body, method=method.upper(), headers=headers)

    try:
        with urllib.request.urlopen(req) as resp:
            raw = resp.read().decode()
            out = {"ok": True, "status": resp.status,
                   "data": json.loads(raw) if raw else {}}
            link = resp.headers.get("Link")
            if link:
                m = re.search(r'page=(\d+)[^>]*>;\s*rel="next"', link)
                if m:
                    out["next_page"] = int(m.group(1))
            rem = resp.headers.get("X-RateLimit-Remaining")
            if rem is not None:
                out["rate_limit_remaining"] = rem
            return out
    except urllib.error.HTTPError as e:
        detail = e.read().decode(errors="replace")
        try:
            detail = json.loads(detail)
        except json.JSONDecodeError:
            pass
        return {"ok": False, "status": e.code, "error": detail,
                "method": method.upper(), "url": url}
    except urllib.error.URLError as e:
        return {"ok": False, "error": f"Network error: {e.reason}", "url": url}


def main() -> None:
    p = argparse.ArgumentParser(description="Freshservice API helper")
    sub = p.add_subparsers(dest="cmd", required=True)

    pr = sub.add_parser("request", help="Call any API v2 endpoint")
    pr.add_argument("method", help="GET, POST, PUT, DELETE")
    pr.add_argument("path", help='API path, e.g. "/tickets" or "/tickets/123"')
    pr.add_argument("--param", action="append", default=[],
                    metavar="KEY=VALUE", help="Query parameter (repeatable)")
    pr.add_argument("--data", help="JSON request body for POST/PUT")

    sub.add_parser("check", help="Verify credentials and connectivity")

    args = p.parse_args()

    if args.cmd == "check":
        res = request("GET", "/agents/me", {}, None)
        if res.get("ok"):
            who = res["data"].get("agent", {})
            print(json.dumps({"ok": True,
                              "connected_as": who.get("email") or who.get("first_name"),
                              "status": res["status"]}, indent=2))
        else:
            print(json.dumps(res, indent=2, default=str))
            sys.exit(1)
        return

    params = {}
    for item in args.param:
        if "=" not in item:
            sys.exit(f"ERROR: --param must be KEY=VALUE, got: {item}")
        k, v = item.split("=", 1)
        params[k] = v

    if args.data:
        try:
            json.loads(args.data)
        except json.JSONDecodeError as e:
            sys.exit(f"ERROR: --data is not valid JSON: {e}")

    res = request(args.method, args.path, params, args.data)
    print(json.dumps(res, indent=2, default=str))
    if not res.get("ok"):
        sys.exit(1)


if __name__ == "__main__":
    main()