ConnectWise PSA skills and tools under ops/connectwise (hub-sanitized 2026-11-01: client intel, internal workflow/credential/member/company ids, archived-page apiToken/sessionId stripped to placeholders)

ops/connectwise/connectwise-msp/scripts/cw_api.py

@@ -0,0 +1,93 @@
+import requests
+import re
+import json
+import os
+import base64
+from requests.auth import HTTPBasicAuth
+
+# Configuration
+BASE_URL = "https://cw.vistanetinc.com/v4_6_release/apis/3.0"
+def _load_cw_client_id():
+    """Resolve the ConnectWise clientId from env or ~/.secrets/connectwise."""
+    cid = os.getenv("CW_CLIENT_ID", "")
+    if cid:
+        return cid
+    path = os.path.expanduser("~/.secrets/connectwise")
+    if os.path.exists(path):
+        with open(path) as f:
+            for line in f:
+                if line.lstrip().startswith("CW_CLIENT_ID="):
+                    return line.split("=", 1)[1].strip()
+    raise SystemExit(
+        "CW_CLIENT_ID missing: export CW_CLIENT_ID or add "
+        "CW_CLIENT_ID=<uuid> to ~/.secrets/connectwise"
+    )
+
+CLIENT_ID = _load_cw_client_id()
+SECRETS_PATH = os.path.expanduser("~/.secrets/connectwise")
+
+def get_auth():
+    with open(SECRETS_PATH, 'r') as f:
+        lines = f.readlines()
+        creds = {}
+        for line in lines:
+            if '=' in line:
+                k, v = line.strip().split('=', 1)
+                creds[k] = v
+    
+    auth_str = f"{creds['CW_COMPANY_ID']}{creds['CW_PUBLIC_KEY']}:{creds['CW_PRIVATE_KEY']}"
+    return auth_str
+
+def clean_json(text):
+    if not text:
+        return text
+    return re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f]', '', text)
+
+def cw_request(method, endpoint, params=None, json_data=None):
+    url = f"{BASE_URL}/{endpoint.lstrip('/')}"
+    auth_str = get_auth()
+    
+    headers = {
+        "clientId": CLIENT_ID,
+        "Content-Type": "application/json",
+        "Accept": "application/json"
+    }
+    
+    encoded = base64.b64encode(auth_str.encode()).decode()
+    headers["Authorization"] = f"Basic {encoded}"
+    
+    response = requests.request(
+        method=method,
+        url=url,
+        params=params,
+        json=json_data,
+        headers=headers,
+        timeout=30
+    )
+    
+    cleaned_text = clean_json(response.text)
+    try:
+        return json.loads(cleaned_text), response.status_code
+    except json.JSONDecodeError:
+        return cleaned_text, response.status_code
+
+def get_ticket(ticket_id):
+    return cw_request("GET", f"service/tickets/{ticket_id}")
+
+def list_tickets(conditions=None, page_size=100):
+    params = {"pageSize": page_size}
+    if conditions:
+        params["conditions"] = conditions
+    return cw_request("GET", "service/tickets", params=params)
+
+def update_ticket(ticket_id, patch_data):
+    return cw_request("PATCH", f"service/tickets/{ticket_id}", json_data=patch_data)
+
+def create_schedule_entry(data):
+    return cw_request("POST", "schedule/entries", json_data=data)
+
+if __name__ == "__main__":
+    print("Testing CW Connection...")
+    data, status = get_ticket(121216)
+    print(f"Status: {status}")
+    print(f"Data: {data}")