@trigger.dev/openai: Adding additional assistant tasks

.changeset/smart-frogs-help.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/openai": patch
+---
+
+Adding additional assistant tasks

apps/webapp/app/env.server.ts

@@ -60,6 +60,7 @@ const EnvironmentSchema = z.object({
   REDIS_PORT: z.coerce.number().optional(),
   REDIS_USERNAME: z.string().optional(),
   REDIS_PASSWORD: z.string().optional(),
+  REDIS_TLS_DISABLED: z.string().optional(),
 
   DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
   DEFAULT_DEV_ENV_EXECUTION_ATTEMPTS: z.coerce.number().int().positive().default(1),

apps/webapp/app/services/runExecutionRateLimiter.server.ts

@@ -397,7 +397,7 @@ function getRateLimiter() {
           username: env.REDIS_USERNAME,
           password: env.REDIS_PASSWORD,
           enableAutoPipelining: true,
-          tls: {}
+          ...(env.REDIS_TLS_DISABLED === "true" ? {} : { tls: {} })
         },
         defaultConcurrency: env.DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT,
       });

docs/integrations/apis/openai/assistants.mdx

@@ -33,6 +33,54 @@ const assistant = await io.openai.beta.assistants.create("create-assistant", {
 });
 ```
 
+### `update()`
+
+Update an assistant. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/assistants/modifyAssistant)
+
+```ts example.ts
+const file = await io.openai.files.createAndWaitForProcessing("upload-file", {
+  purpose: "assistants",
+  file: fs.createReadStream("./fixtures/mydata.csv"),
+});
+
+const assistantId = "asst_abc123";
+
+const assistant = await io.openai.beta.assistants.update("update-assistant", assistantId, {
+  file_ids: [file.id], // add a file to the assistant
+});
+```
+
+### `list()`
+
+List assistants. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/assistants/listAssistants)
+
+```ts example.ts
+const assistants = await io.openai.beta.assistants.list("list");
+
+// with pagination
+const assistants = await io.openai.beta.assistants.list("list", {
+  limit: 10,
+  order: "desc",
+  after: "asst_abc123",
+});
+```
+
+### `retrieve()`
+
+Retrieve an assistant. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/assistants/getAssistant)
+
+```ts example.ts
+const assistant = await io.openai.beta.assistants.retrieve("get-assistant", "asst_abc123");
+```
+
+### `del()`
+
+Delete an assistant. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/assistants/deleteAssistant)
+
+```ts example.ts
+const deletedAssistant = await io.openai.beta.assistants.del("delete-assistant", "asst_abc123");
+```
+
 ## Threads
 
 Create threads that assistants can interact with. [Official OpenAI docs](https://platform.openai.com/docs/api-reference/threads/createThread)
@@ -132,12 +180,26 @@ Create messages within threads. [Official OpenAI docs](https://platform.openai.c
 
 ### `list()`
 
-List all messages in a thread.
+List messages in a thread.
 
 ```ts example.ts
 const messages = await io.openai.beta.threads.messages.list("list-messages", "thread_abc123");
+// with pagination
+const messages = await io.openai.beta.threads.messages.list("list-messages", "thread_abc123", {
+  limit: 10,
+  order: "desc",
+  after: "message_abc123",
+});
 ```
 
+If you want to list all messages in a thread, you can use the `listAll()` helper:
+
+```ts example.ts
+const messages = await io.openai.beta.threads.messages.listAll("list-messages", "thread_abc123");
+```
+
+This will automatically paginate through all messages in the thread and return them as a single array.
+
 ### `create()`
 
 Create a message. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/messages/createMessage)

integrations/openai/src/assistants.ts

@@ -2,13 +2,13 @@ import { IntegrationTaskKey, Prettify } from "@trigger.dev/sdk";
 import { OpenAIRunTask } from "./index";
 import { OpenAIIntegrationOptions, OpenAIRequestOptions } from "./types";
 import OpenAI from "openai";
-import { createTaskOutputProperties, handleOpenAIError } from "./taskUtils";
+import { createTaskOutputProperties, handleOpenAIError, isRequestOptions } from "./taskUtils";
 
 export class Assistants {
   constructor(
     private runTask: OpenAIRunTask,
     private options: OpenAIIntegrationOptions
-  ) {}
+  ) { }
 
   async create(
     key: IntegrationTaskKey,
@@ -54,4 +54,173 @@ export class Assistants {
       handleOpenAIError
     );
   }
+
+  async update(
+    key: IntegrationTaskKey,
+    id: string,
+    params: Prettify<OpenAI.Beta.AssistantUpdateParams>,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Assistant> {
+    return this.runTask(
+      key,
+      async (client, task) => {
+        const { data, response } = await client.beta.assistants
+          .update(id, params, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        const outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        task.outputProperties = [
+          ...(outputProperties ?? []),
+          {
+            label: "assistantId",
+            text: data.id,
+          },
+        ];
+
+        return data;
+      },
+      {
+        name: "Update Assistant",
+        params,
+        properties: [
+          ...(params.model ? [{ label: "model", text: params.model }] : []),
+          ...(params.name ? [{ label: "name", text: params.name }] : []),
+          ...(params.file_ids && params.file_ids.length > 0
+            ? [{ label: "files", text: params.file_ids.join(", ") }]
+            : []),
+        ],
+      },
+      handleOpenAIError
+    );
+  }
+
+  list(
+    key: IntegrationTaskKey,
+    params?: Prettify<OpenAI.Beta.AssistantListParams>,
+    options?: OpenAIRequestOptions,
+  ): Promise<OpenAI.Beta.Assistant[]>;
+  list(
+    key: IntegrationTaskKey,
+    options?: OpenAIRequestOptions,
+  ): Promise<OpenAI.Beta.Assistant[]>;
+  async list(
+    key: IntegrationTaskKey,
+    params: Prettify<OpenAI.Beta.AssistantListParams> | OpenAIRequestOptions = {},
+    options: OpenAIRequestOptions | undefined = undefined
+  ): Promise<OpenAI.Beta.Assistant[]> {
+    return this.runTask(
+      key,
+      async (client, task) => {
+        if (isRequestOptions(params)) {
+          const { data, response } = await client.beta.assistants
+            .list({
+              idempotencyKey: task.idempotencyKey,
+              ...params,
+            })
+            .withResponse();
+
+          task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+          return data.data;
+        }
+
+        const { data, response } = await client.beta.assistants
+          .list(params, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return data.data;
+
+      },
+      {
+        name: "List Assistants",
+        params,
+        properties: !isRequestOptions(params) ? [
+          ...(params.before ? [{ label: "before", text: params.before }] : []),
+          ...(params.order ? [{ label: "order", text: params.order }] : []),
+          ...(params.after ? [{ label: "after", text: params.after }] : []),
+          ...(params.limit ? [{ label: "limit", text: String(params.limit) }] : []),
+        ] : [],
+      },
+      handleOpenAIError
+    );
+  }
+
+  async del(
+    key: IntegrationTaskKey,
+    id: string,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.AssistantDeleted> {
+    return this.runTask(
+      key,
+      async (client, task) => {
+        const { data, response } = await client.beta.assistants
+          .del(id, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return data;
+      },
+      {
+        name: "Delete Assistant",
+        params: {
+          id,
+        },
+        properties: [
+          {
+            label: "assistantId",
+            text: id,
+          },
+        ],
+      },
+      handleOpenAIError
+    );
+  }
+
+  async retrieve(
+    key: IntegrationTaskKey,
+    id: string,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Assistant> {
+    return this.runTask(
+      key,
+      async (client, task) => {
+        const { data, response } = await client.beta.assistants
+          .retrieve(id, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return data;
+      },
+      {
+        name: "Retrieve Assistant",
+        params: {
+          id,
+        },
+        properties: [
+          {
+            label: "assistantId",
+            text: id,
+          },
+        ],
+      },
+      handleOpenAIError
+    );
+  }
 }

integrations/openai/src/taskUtils.ts

@@ -60,11 +60,11 @@ function createTaskUsageProperties(
     },
     ...("completion_tokens" in usage
       ? [
-          {
+        {
-            label: "Completion Usage",
+          label: "Completion Usage",
-            text: String(usage.completion_tokens),
+          text: String(usage.completion_tokens),
-          },
+        },
-        ]
+      ]
       : []),
   ];
 }
@@ -83,35 +83,35 @@ function createTaskRateLimitProperties(headers: Headers | undefined) {
   return [
     ...(remainingRequests
       ? [
-          {
+        {
-            label: "Remaining Requests",
+          label: "Remaining Requests",
-            text: remainingRequests ?? "Unknown",
+          text: remainingRequests ?? "Unknown",
-          },
+        },
-        ]
+      ]
       : []),
     ...(resetRequests
       ? [
-          {
+        {
-            label: "Reset Requests",
+          label: "Reset Requests",
-            text: resetRequests ?? "Unknown",
+          text: resetRequests ?? "Unknown",
-          },
+        },
-        ]
+      ]
       : []),
     ...(remainingTokens
       ? [
-          {
+        {
-            label: "Remaining Tokens",
+          label: "Remaining Tokens",
-            text: remainingTokens ?? "Unknown",
+          text: remainingTokens ?? "Unknown",
-          },
+        },
-        ]
+      ]
       : []),
     ...(resetTokens
       ? [
-          {
+        {
-            label: "Reset Tokens",
+          label: "Reset Tokens",
-            text: resetTokens ?? "Unknown",
+          text: resetTokens ?? "Unknown",
-          },
+        },
-        ]
+      ]
       : []),
   ];
 }
@@ -282,3 +282,32 @@ export const backgroundTaskRetries: FetchRetryOptions = {
     randomize: true,
   },
 };
+
+type KeysEnum<T> = { [P in keyof Required<T>]: true };
+
+const requestOptionsKeys: KeysEnum<OpenAIRequestOptions> = {
+  method: true,
+  path: true,
+  query: true,
+  headers: true,
+  idempotencyKey: true,
+};
+
+export const isRequestOptions = (obj: unknown): obj is OpenAIRequestOptions => {
+  return (
+    typeof obj === 'object' &&
+    obj !== null &&
+    !isEmptyObj(obj) &&
+    Object.keys(obj).every((k) => hasOwn(requestOptionsKeys, k))
+  );
+};
+
+function isEmptyObj(obj: Object | null | undefined): boolean {
+  if (!obj) return true;
+  for (const _k in obj) return false;
+  return true;
+}
+
+function hasOwn(obj: Object, key: string): boolean {
+  return Object.prototype.hasOwnProperty.call(obj, key);
+}

integrations/openai/src/threads.ts

@@ -7,6 +7,7 @@ import {
   createBackgroundFetchUrl,
   createTaskOutputProperties,
   handleOpenAIError,
+  isRequestOptions,
 } from "./taskUtils";
 import { RunSubmitToolOutputsParams } from "openai/resources/beta/threads/runs/runs";
 import { ThreadUpdateParams } from "openai/resources/beta/threads/threads";
@@ -15,7 +16,7 @@ export class Threads {
   constructor(
     private runTask: OpenAIRunTask,
     private options: OpenAIIntegrationOptions
-  ) {}
+  ) { }
 
   /**
    * Create a thread and run it in one task.
@@ -261,7 +262,7 @@ class Runs {
   constructor(
     private runTask: OpenAIRunTask,
     private options: OpenAIIntegrationOptions
-  ) {}
+  ) { }
 
   /**
    * Creates a run and waits for it to complete by polling in the background.
@@ -551,15 +552,70 @@ class Messages {
   constructor(
     private runTask: OpenAIRunTask,
     private options: OpenAIIntegrationOptions
-  ) {}
+  ) { }
+
+  /**
+   * Returns messages for a given thread.
+   */
+  list(
+    key: IntegrationTaskKey,
+    threadId: string,
+    params?: Prettify<OpenAI.Beta.Threads.MessageListParams>,
+    options?: OpenAIRequestOptions
+  ): Promise<OpenAI.Beta.Threads.ThreadMessage[]>
+  list(
+    key: IntegrationTaskKey,
+    threadId: string,
+    options?: OpenAIRequestOptions
+  ): Promise<OpenAI.Beta.Threads.ThreadMessage[]>
+  async list(
+    key: IntegrationTaskKey,
+    threadId: string,
+    params: Prettify<OpenAI.Beta.AssistantListParams> | OpenAIRequestOptions = {},
+    options: OpenAIRequestOptions | undefined = undefined
+  ): Promise<OpenAI.Beta.Threads.ThreadMessage[]> {
+    return this.runTask(
+      key,
+      async (client, task, io) => {
+        if (isRequestOptions(params)) {
+          const { data: page, response } = await client.beta.threads.messages
+            .list(threadId, {
+              idempotencyKey: task.idempotencyKey,
+              ...params,
+            })
+            .withResponse();
+
+          task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+          return page.data;
+        }
+
+        const { data: page, response } = await client.beta.threads.messages
+          .list(threadId, params, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return page.data;
+      },
+      {
+        name: "List Messages",
+        properties: [{ label: "threadId", text: threadId }],
+      },
+      handleOpenAIError
+    );
+  }
 
   /**
    * Returns all messages for a given thread.
    */
-  async list(
+  async listAll(
     key: IntegrationTaskKey,
     threadId: string,
-    options: OpenAIRequestOptions = {}
+    options: OpenAIRequestOptions = {},
   ): Promise<OpenAI.Beta.Threads.ThreadMessage[]> {
     return this.runTask(
       key,
@@ -573,8 +629,8 @@ class Messages {
 
         const allMessages = [];
 
-        for await (const fineTuningJob of page) {
+        for await (const message of page) {
-          allMessages.push(fineTuningJob);
+          allMessages.push(message);
         }
 
         task.outputProperties = createTaskOutputProperties(undefined, response.headers);
@@ -582,7 +638,7 @@ class Messages {
         return allMessages;
       },
       {
-        name: "List Messages",
+        name: "List All Messages",
         properties: [{ label: "threadId", text: threadId }],
       },
       handleOpenAIError

integrations/openai/src/types.ts

@@ -34,4 +34,4 @@ export type OpenAIRequestOptions = {
   path?: string;
   headers?: OpenAIHeaders;
   idempotencyKey?: string;
-};
+};

references/job-catalog/src/openai.ts

@@ -228,6 +228,37 @@ client.defineJob({
   },
 });
 
+client.defineJob({
+  id: "openai-manage-assistant",
+  name: "OpenAI GPT Manage Assistant",
+  version: "0.0.1",
+  trigger: invokeTrigger({
+    schema: z.object({
+      assistantId: z.string().optional(),
+    }),
+  }),
+  integrations: {
+    openai,
+  },
+  run: async (payload, io, ctx) => {
+    const assistants = await io.openai.beta.assistants.list("list", {
+      limit: 10
+    });
+
+    if (payload.assistantId) {
+      await io.openai.beta.assistants.retrieve("retrieve", payload.assistantId);
+      await io.openai.beta.assistants.update("update", payload.assistantId, {
+        name: "Updated name",
+      });
+      await io.openai.beta.assistants.del("delete", payload.assistantId);
+    }
+
+    for (const assistant of assistants) {
+      await io.openai.beta.assistants.del(`delete ${assistant.id}`, assistant.id);
+    }
+  },
+});
+
 client.defineJob({
   id: "openai-use-assistant",
   name: "OpenAI GPT Use Assistant",