Improve the script to support approvals

step-templates/octopus-ai-prompt.json

@@ -3,15 +3,15 @@
   "Name": "Octopus - Prompt AI",
   "Description": "Prompt the Octopus AI service with a message and store the result in a variable. See https://octopus.com/docs/administration/copilot for more information.",
   "ActionType": "Octopus.Script",
-  "Version": 1,
+  "Version": 2,
   "CommunityActionTemplateId": null,
   "Packages": [],
   "GitDependencies": [],
   "Properties": {
     "Octopus.Action.RunOnServer": "true",
     "Octopus.Action.Script.ScriptSource": "Inline",
     "Octopus.Action.Script.Syntax": "Python",
-    "Octopus.Action.Script.ScriptBody": "import os\nimport re\nimport http.client\nimport json\n\n# If this script is not being run as part of an Octopus step, return variables from environment variables.\n# Periods are replaced with underscores, and the variable name is converted to uppercase\nif 'get_octopusvariable' not in globals():\n    def get_octopusvariable(variable):\n        return os.environ.get(re.sub('\\\\.', '_', variable.upper()))\n\nif 'set_octopusvariable' not in globals():\n    def set_octopusvariable(variable, value):\n        print(f\"Setting {variable} to {value}\")\n\n# If this script is not being run as part of an Octopus step, print directly to std out.\nif 'printverbose' not in globals():\n    def printverbose(msg):\n        print(msg)\n\ndef make_post_request(message, github_token, octopus_api_key, octopus_server):\n    \"\"\"\n    Query the Octopus AI service with a message.\n    :param message: The prompt message\n    :param github_token: The GitHub token\n    :param octopus_api_key: The Octopus API key\n    :param octopus_server: The Octopus URL\n    :return: The AI response\n    \"\"\"\n    headers = {\n        \"X-GitHub-Token\": github_token,\n        \"X-Octopus-ApiKey\": octopus_api_key,\n        \"X-Octopus-Server\": octopus_server,\n        \"Content-Type\": \"application/json\"\n    }\n    body = json.dumps({\"messages\": [{\"content\": message}]}).encode(\"utf8\")\n\n    conn = http.client.HTTPSConnection(\"aiagent.octopus.com\")\n    conn.request(\"POST\", \"/api/form_handler\", body, headers)\n    response = conn.getresponse()\n    response_data = response.read().decode(\"utf8\")\n    conn.close()\n\n    return convert_from_sse_response(response_data)\n\n\ndef convert_from_sse_response(sse_response):\n    \"\"\"\n    Converts an SSE response into a string.\n    :param sse_response: The SSE response to convert.\n    :return: The string representation of the SSE response.\n    \"\"\"\n\n    responses = map(\n        lambda line: json.loads(line.replace(\"data: \", \"\")),\n        filter(lambda line: line.strip(), sse_response.split(\"\\n\")),\n    )\n    content_responses = filter(\n        lambda response: \"content\" in response[\"choices\"][0][\"delta\"], responses\n    )\n    return \"\\n\".join(\n        map(\n            lambda line: line[\"choices\"][0][\"delta\"][\"content\"].strip(),\n            content_responses,\n        )\n    )\n\nstep_name = get_octopusvariable(\"Octopus.Step.Name\")\nmessage = get_octopusvariable(\"OctopusAI.Prompt\")\ngithub_token = get_octopusvariable(\"OctopusAI.GitHub.Token\")\noctopus_api = get_octopusvariable(\"OctopusAI.Octopus.APIKey\")\noctopus_url = get_octopusvariable(\"OctopusAI.Octopus.Url\")\n\nresult = make_post_request(message, github_token, octopus_api, octopus_url)\n\nset_octopusvariable(\"AIResult\", result)\n\nprint(result)\nprint(f\"AI result is available in the variable: Octopus.Action[{step_name}].Output.AIResult\")"
+    "Octopus.Action.Script.ScriptBody": "import os\nimport re\nimport http.client\nimport json\nimport time\nfrom urllib.parse import quote\n\n# If this script is not being run as part of an Octopus step, return variables from environment variables.\n# Periods are replaced with underscores, and the variable name is converted to uppercase\nif 'get_octopusvariable' not in globals():\n    def get_octopusvariable(variable):\n        return os.environ.get(re.sub('\\\\.', '_', variable.upper()))\n\nif 'set_octopusvariable' not in globals():\n    def set_octopusvariable(variable, value):\n        print(f\"Setting {variable} to {value}\")\n\n# If this script is not being run as part of an Octopus step, print directly to std out.\nif 'printverbose' not in globals():\n    def printverbose(msg):\n        print(msg)\n\ndef make_post_request(message, auto_approve, github_token, octopus_api_key, octopus_server, retry = 0):\n    \"\"\"\n    Query the Octopus AI service with a message.\n    :param message: The prompt message\n    :param github_token: The GitHub token\n    :param octopus_api_key: The Octopus API key\n    :param octopus_server: The Octopus URL\n    :return: The AI response\n    \"\"\"\n    headers = {\n        \"X-GitHub-Token\": github_token,\n        \"X-Octopus-ApiKey\": octopus_api_key,\n        \"X-Octopus-Server\": octopus_server,\n        \"Content-Type\": \"application/json\"\n    }\n    body = json.dumps({\"messages\": [{\"content\": message}]}).encode(\"utf8\")\n\n    conn = http.client.HTTPSConnection(\"aiagent.octopus.com\", timeout=240)\n    conn.request(\"POST\", \"/api/form_handler\", body, headers)\n    response = conn.getresponse()\n    response_data = response.read().decode(\"utf8\")\n    conn.close()\n\n    if response.status < 200 or response.status > 300:\n        if retry < 2:\n            printverbose(f\"Request to AI Agent failed with status code {response.status} and message: {response.reason}. Retrying...\")\n            time.sleep(400)\n            return make_post_request(message, auto_approve, github_token, octopus_api_key, octopus_server, retry + 1)\n        return f\"Request to AI Agent failed with status code {response.status} and message: {response.reason}\"\n\n    if is_action_response(response_data):\n        if auto_approve:\n            id = action_response_id(response_data)\n\n            if not id:\n                return \"Prompt required approval, but no confirmation ID was found in the response.\"\n\n            conn = http.client.HTTPSConnection(\"aiagent.octopus.com\", timeout=240)\n            conn.request(\"POST\", \"/api/form_handler?confirmation_id=\" + quote(id, safe='') + \"&confirmation_state=accepted\", body, headers)\n            response = conn.getresponse()\n            response_data = response.read().decode(\"utf8\")\n            conn.close()\n            return convert_from_sse_response(response_data)\n        else:\n            return \"Prompt required approval, but auto-approval is disabled. Please enable the auto-approval option in the step.\"\n\n    return convert_from_sse_response(response_data)\n\n\ndef convert_from_sse_response(sse_response):\n    \"\"\"\n    Converts an SSE response into a string.\n    :param sse_response: The SSE response to convert.\n    :return: The string representation of the SSE response.\n    \"\"\"\n\n    responses = map(\n        lambda line: json.loads(line.replace(\"data: \", \"\")),\n        filter(lambda line: line.strip().startswith(\"data:\"), sse_response.split(\"\\n\")),\n    )\n    content_responses = filter(\n        lambda response: \"content\" in response[\"choices\"][0][\"delta\"], responses\n    )\n    return \"\\n\".join(\n        map(\n            lambda line: line[\"choices\"][0][\"delta\"][\"content\"].strip(),\n            content_responses,\n        )\n    )\n\ndef is_action_response(sse_response):\n    responses = map(\n        lambda line: json.loads(line.replace(\"data: \", \"\")),\n        filter(lambda line: line.strip().startswith(\"data:\"), sse_response.split(\"\\n\")),\n    )\n\n    return any(response.get(\"type\") == \"action\" for response in responses)\n\ndef action_response_id(sse_response):\n    responses = map(\n        lambda line: json.loads(line.replace(\"data: \", \"\")),\n        filter(lambda line: line.strip().startswith(\"data:\"), sse_response.split(\"\\n\")),\n    )\n\n    action = next(filter(lambda response: response.get(\"type\") == \"action\", responses))\n\n    return action.get(\"confirmation\", {}).get(\"id\")\n\nstep_name = get_octopusvariable(\"Octopus.Step.Name\")\nmessage = get_octopusvariable(\"OctopusAI.Prompt\")\ngithub_token = get_octopusvariable(\"OctopusAI.GitHub.Token\")\noctopus_api = get_octopusvariable(\"OctopusAI.Octopus.APIKey\")\noctopus_url = get_octopusvariable(\"OctopusAI.Octopus.Url\")\nauto_approve = get_octopusvariable(\"OctopusAI.AutoApprove\").casefold() == \"true\"\n\nresult = make_post_request(message, auto_approve, github_token, octopus_api, octopus_url)\n\nset_octopusvariable(\"AIResult\", result)\n\nprint(result)\nprint(f\"AI result is available in the variable: Octopus.Action[{step_name}].Output.AIResult\")\n"
   },
   "Parameters": [
     {
@@ -53,6 +53,16 @@
       "DisplaySettings": {
         "Octopus.ControlType": "SingleLineText"
       }
+    },
+    {
+      "Id": "380c09c0-2af8-4866-a1a7-db77fa89f7d8",
+      "Name": "OctopusAI.AutoApprove",
+      "Label": "Enable this to auto approve any prompts that require approval",
+      "HelpText": null,
+      "DefaultValue": "False",
+      "DisplaySettings": {
+        "Octopus.ControlType": "Checkbox"
+      }
     }
   ],
   "StepPackageId": "Octopus.Script",