#!/usr/bin/env python3 """GenHealth synthetic workflow quickstart. Python 3; no third-party packages. validate DEFINITION: preflight only. execute DEFINITION: publish one synthetic run. inspect: observe the run in --state. No external integration actions are executed. Run state contains no key and must be retained to resume after an interruption. Use one process per state file. Review this client before running it. """ import argparse import hashlib import json import os from pathlib import Path import sys from urllib.error import HTTPError, URLError from urllib.parse import urlencode, urlsplit from urllib.request import HTTPRedirectHandler, Request, build_opener from uuid import UUID, uuid4 class ApiError(RuntimeError): """A bounded diagnostic that does not echo response bodies or credentials.""" class NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): raise ApiError("API redirects are refused; verify GENHEALTH_API_BASE.") class Client: def __init__(self, base_url, api_key): parsed = urlsplit(base_url) local_http = parsed.scheme == "http" and parsed.hostname in {"localhost", "127.0.0.1", "::1"} if (parsed.scheme != "https" and not local_http) or not parsed.hostname: raise ApiError("Use an HTTPS API origin, or HTTP localhost for an isolated local server.") if parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"}: raise ApiError("GENHEALTH_API_BASE must be an origin without credentials, path, query, or fragment.") if not api_key: raise ApiError("Set GENHEALTH_API_KEY to an organization-bound key.") self.base_url = base_url.rstrip("/") self.api_key = api_key self.opener = build_opener(NoRedirect()) def request(self, method, path, body=None): request = Request( self.base_url + "/developer/v1" + path, data=json.dumps(body).encode() if body is not None else None, headers={"Authorization": "Bearer " + self.api_key, "Content-Type": "application/json", "Accept": "application/json"}, method=method, ) try: with self.opener.open(request, timeout=30) as response: return json.load(response) except HTTPError as error: raise ApiError("HTTP %s: consult /docs/workflows/operations; no automatic write retry was attempted." % error.code) from None except (URLError, TimeoutError, OSError): raise ApiError("Transport outcome unknown. Retain the state file and resume using the documented recovery procedure.") from None def save_state(path, state): temporary = path.with_suffix(path.suffix + ".tmp") # Owner-only permissions from creation; state is synthetic but operationally sensitive. descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) with os.fdopen(descriptor, "w") as stream: json.dump(state, stream, indent=2) temporary.replace(path) def preflight(client, definition): result = client.request("POST", "/workflows/validate", {"definition": definition}) if result.get("viable") is not True or result.get("readiness") != "ready": raise ApiError("Preflight is blocked or needs review. Inspect validation findings through the authenticated API before publishing.") return result def execute(client, definition, path): # This executable example is a smoke test, not an arbitrary production runner. nodes = definition.get("workflow", {}).get("nodes", []) if not nodes or any(node.get("type") not in {"comment", "if"} for node in nodes): raise ApiError("This client executes synthetic comment/if graphs only. Use the documented API lifecycle for integrations.") fingerprint = hashlib.sha256(json.dumps(definition, sort_keys=True).encode()).hexdigest() state = json.loads(path.read_text()) if path.exists() else {"stage": "new", "base_url": client.base_url, "definition_sha256": fingerprint} if state.get("base_url") != client.base_url or state.get("definition_sha256") != fingerprint: raise ApiError("State belongs to a different API origin or definition; do not reuse it.") preflight(client, definition) if state["stage"] == "import_pending": raise ApiError("Import outcome unknown. List workflows and verify the candidate with GET /workflows/{id}/export. After confirming its graph matches this definition, set workflow_id and stage=imported in the state file, then resume. Do not repeat import with a new state file.") if state["stage"] == "new": state["stage"] = "import_pending" save_state(path, state) imported = client.request("POST", "/workflows/import", definition) state.update(workflow_id=str(UUID(imported["workflow"]["id"])), stage="imported") save_state(path, state) workflow_path = "/workflows/" + str(UUID(state["workflow_id"])) if state["stage"] == "draft_pending": current = client.request("GET", workflow_path) if not current.get("has_draft"): raise ApiError("Draft outcome unknown; inspect workflow versions before retrying a write.") state["stage"] = "drafted" save_state(path, state) if state["stage"] == "imported": state["stage"] = "draft_pending" save_state(path, state) client.request("POST", workflow_path + "/draft") state["stage"] = "drafted" save_state(path, state) if state["stage"] == "publish_pending": current = client.request("GET", workflow_path) if current.get("status") != "active" or current.get("has_draft"): raise ApiError("Publication outcome unknown; inspect workflow versions before retrying a write.") state["stage"] = "published" save_state(path, state) if state["stage"] == "drafted": state["stage"] = "publish_pending" save_state(path, state) client.request("POST", workflow_path + "/publish", {}) state["stage"] = "published" save_state(path, state) if state["stage"] == "published": current = client.request("GET", workflow_path) state.update(stage="run_pending", version_id=current["active_version_id"], run_request={"idempotency_key": str(uuid4()), "trigger_data": {"ready": True}}) save_state(path, state) if state["stage"] == "run_pending": current = client.request("GET", workflow_path) if current.get("active_version_id") != state["version_id"]: raise ApiError("Active version changed. Reconcile the pending run; do not change its idempotency key.") run = client.request("POST", workflow_path + "/runs", state["run_request"]) state.update(stage="accepted", run_id=str(UUID(run["run_id"]))) save_state(path, state) if state["stage"] != "accepted": raise ApiError("Unknown state stage; inspect the state file before another write.") return state def inspect(client, state): if state.get("base_url") != client.base_url or not state.get("run_id"): raise ApiError("No accepted run for this API origin. Resume execute first.") workflow_id, run_id = str(UUID(state["workflow_id"])), str(UUID(state["run_id"])) result = client.request("GET", "/workflows/" + workflow_id + "/runs?" + urlencode({"limit": 25})) run = next((item for item in result.get("runs", []) if item.get("run_id") == run_id), None) if run is None: raise ApiError("Run not in the returned page. Use since/until/status filters in the API; do not submit another run.") print(json.dumps({"workflow_id": workflow_id, "run_id": run_id, "status": run.get("status")}, indent=2)) if run.get("status") in {"failed", "cancelled"}: raise ApiError("Run ended without completion; inspect node shapes in the authenticated API.") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("command", choices=["validate", "execute", "inspect"]) parser.add_argument("definition", nargs="?") parser.add_argument("--state", type=Path, default=Path("agent-smoke-state.json")) args = parser.parse_args() client = Client(os.environ.get("GENHEALTH_API_BASE", "https://api.umpa.genhealth.ai"), os.environ.get("GENHEALTH_API_KEY")) if args.command == "inspect": inspect(client, json.loads(args.state.read_text())) return if not args.definition: parser.error("validate and execute require a definition file") definition = json.loads(Path(args.definition).read_text()) if args.command == "validate": preflight(client, definition) print("Preflight ready; no workflow was created.") else: state = execute(client, definition, args.state) print(json.dumps({"workflow_id": state["workflow_id"], "run_id": state["run_id"], "status": "accepted; inspect for completion"}, indent=2)) if __name__ == "__main__": try: main() except (ApiError, OSError, ValueError, KeyError, TypeError) as error: # JSON/OS diagnostics can contain file contents or paths; only ApiError is curated. print(str(error) if isinstance(error, ApiError) else "Invalid local state, definition, or API response; inspect it locally before continuing.", file=sys.stderr) sys.exit(1)