feat(sdk): add agents.run() and AgentStream.threadId

seanphan · seanphan · commit 5196fb974d11 · 2026-04-02T00:05:05.000-07:00
SDK:
- AgentStream now extracts threadId and userMessageId from the
  thread_info data event during streaming
- New agents.run(agentId, { message, threadId?, timeoutMs? }) method:
  sends a task, waits for result, returns { response, threadId, status }
- Internally streams to collect text + threadId, falls back to polling
  GET /agent-threads/{id}/messages if stream returns empty
- Export AgentRunResult, AgentRunOptions types

CLI:
- `af agent run` now delegates to sdk agents.run() (was 80 lines inline,
  now 20 lines)
- Returns structured JSON: { schema, status, agent_id, thread_id, response }

Usage (AI agent calling via Bash):
  af agent run --agent-id &lt;id&gt; --message "Analyze X" --json
  → {"status":"completed","thread_id":"...","response":"..."}
diff --git a/packages/cli/src/cli/main.ts b/packages/cli/src/cli/main.ts
@@ -3623,90 +3623,35 @@ export function createProgram(): Command {
     .option("--poll-interval <seconds>", "Seconds between status checks", "2")
     .action(async (opts) => {
       const client = buildClient(program.opts());
-      const timeoutMs = Number.parseInt(opts.timeout, 10) * 1000;
-      const pollMs = Number.parseInt(opts.pollInterval, 10) * 1000;
-      const threadId = opts.threadId ?? crypto.randomUUID();
 
-      try {
-        // 1. Fire: POST to stream, read thread_id from first line, don't block
-        if (!isJsonFlagEnabled()) {
-          console.error(`Sending task to agent ${opts.agentId}...`);
-        }
-
-        const streamBody = {
-          messages: [{ role: "user" as const, content: opts.message }],
-        };
-
-        const token = resolveToken(program.opts());
-        const stream = token
-          ? await client.agents.stream(opts.agentId, streamBody)
-          : await client.agents.streamAnonymous(opts.agentId, streamBody);
-
-        // Consume stream to get thread info and full text
-        const text = await stream.text();
-
-        // Try to get thread_id from stream metadata
-        let resolvedThreadId = threadId;
-        try {
-          const rawParts = (stream as unknown as { _raw?: string })._raw;
-          if (typeof rawParts === "string") {
-            for (const line of rawParts.split("\n")) {
-              if (line.startsWith("2:")) {
-                const arr = JSON.parse(line.slice(2)) as Array<{ type: string; data: Record<string, unknown> }>;
-                const info = arr.find((e) => e.type === "thread_info");
-                if (info?.data?.thread_id) {
-                  resolvedThreadId = info.data.thread_id as string;
-                }
-              }
-            }
-          }
-        } catch { /* ignore — use default threadId */ }
+      if (!isJsonFlagEnabled()) {
+        console.error(`Sending task to agent ${opts.agentId}...`);
+      }
 
-        // 2. If we got text from stream, return it immediately
-        if (text && text.trim()) {
-          printResult({
-            schema: "agenticflow.agent.run.v1",
-            status: "completed",
-            agent_id: opts.agentId,
-            thread_id: resolvedThreadId,
-            response: text,
-          });
-          return;
-        }
+      try {
+        const result = await client.agents.run(opts.agentId, {
+          message: opts.message,
+          threadId: opts.threadId,
+          timeoutMs: Number.parseInt(opts.timeout, 10) * 1000,
+          pollIntervalMs: Number.parseInt(opts.pollInterval, 10) * 1000,
+        });
 
-        // 3. Fallback: poll thread status then fetch messages
-        if (!isJsonFlagEnabled()) {
-          console.error(`Waiting for response (thread: ${resolvedThreadId})...`);
+        if (result.status === "failed") {
+          fail("agent_run_failed", `Agent run failed (thread: ${result.threadId})`);
         }
-
-        const start = Date.now();
-        while (Date.now() - start < timeoutMs) {
-          try {
-            const thread = (await client.agentThreads.get(resolvedThreadId)) as Record<string, unknown>;
-            const status = thread.status as string;
-            if (status === "processed" || status === "idle") {
-              const history = (await client.agentThreads.getMessages(resolvedThreadId)) as {
-                messages: Array<{ role: string; content: string }>;
-              };
-              const msgs = history.messages?.filter((m) => m.role === "assistant") ?? [];
-              const lastMsg = msgs.length > 0 ? msgs[msgs.length - 1].content : "";
-              printResult({
-                schema: "agenticflow.agent.run.v1",
-                status: "completed",
-                agent_id: opts.agentId,
-                thread_id: resolvedThreadId,
-                response: lastMsg,
-              });
-              return;
-            }
-            if (status === "failed") {
-              fail("agent_run_failed", `Agent run failed (thread: ${resolvedThreadId})`);
-            }
-          } catch { /* thread not ready yet */ }
-          await new Promise((r) => setTimeout(r, pollMs));
+        if (result.status === "timeout") {
+          fail("agent_run_timeout",
+            `Agent did not respond within ${opts.timeout}s`,
+            `Thread: ${result.threadId}. Check with: af agent-threads messages --thread-id ${result.threadId}`);
         }
 
-        fail("agent_run_timeout", `Agent did not respond within ${opts.timeout}s`, `Thread: ${resolvedThreadId}. Check with: af agent-threads messages --thread-id ${resolvedThreadId}`);
+        printResult({
+          schema: "agenticflow.agent.run.v1",
+          status: result.status,
+          agent_id: opts.agentId,
+          thread_id: result.threadId,
+          response: result.response,
+        });
       } catch (err) {
         const message = err instanceof Error ? err.message : String(err);
         fail("agent_run_failed", message);
diff --git a/packages/sdk/src/index.ts b/packages/sdk/src/index.ts
@@ -23,6 +23,8 @@ export {
 } from "./exceptions.js";
 export {
   AgentsResource,
+  type AgentRunResult,
+  type AgentRunOptions,
   WorkflowsResource,
   ConnectionsResource,
   NodeTypesResource,
diff --git a/packages/sdk/src/resources/agents.ts b/packages/sdk/src/resources/agents.ts
@@ -1,6 +1,28 @@
 import type { AgenticFlowSDK } from "../core.js";
 import { AgentStream, type StreamRequest } from "../streaming.js";
 
+/** Result of a non-streaming agent run. */
+export interface AgentRunResult {
+  /** Agent response text. */
+  response: string;
+  /** Thread ID for conversation continuity. */
+  threadId: string;
+  /** "completed" | "timeout" | "failed" */
+  status: string;
+}
+
+/** Options for `agents.run()`. */
+export interface AgentRunOptions {
+  /** Message to send. */
+  message: string;
+  /** Thread ID to continue a conversation. Auto-generated if omitted. */
+  threadId?: string;
+  /** Max milliseconds to wait. Default 120000 (2 min). */
+  timeoutMs?: number;
+  /** Milliseconds between poll attempts. Default 2000. */
+  pollIntervalMs?: number;
+}
+
 export class AgentsResource {
   constructor(private client: AgenticFlowSDK) { }
 
@@ -63,6 +85,70 @@ export class AgentsResource {
     return new AgentStream(response);
   }
 
+  // ── Run (non-streaming, fire → collect → return) ───────────────────
+
+  /**
+   * Send a message to an agent and return the full response.
+   * Non-streaming — blocks until the agent finishes, then returns text.
+   *
+   * Ideal for AI agents calling the SDK programmatically:
+   * ```ts
+   * const result = await client.agents.run("agent-id", { message: "Analyze this" });
+   * console.log(result.response);   // agent's answer
+   * console.log(result.threadId);   // for follow-up
+   * ```
+   *
+   * Internally: streams to get thread_id + text. If stream returns empty,
+   * falls back to polling GET /agent-threads/{id}/messages.
+   */
+  async run(agentId: string, options: AgentRunOptions): Promise<AgentRunResult> {
+    const threadId = options.threadId ?? crypto.randomUUID();
+    const timeoutMs = options.timeoutMs ?? 120_000;
+    const pollIntervalMs = options.pollIntervalMs ?? 2_000;
+
+    const streamReq: StreamRequest = {
+      id: threadId,
+      messages: [{ role: "user", content: options.message }],
+    };
+
+    // 1. Stream to get response text + thread_id
+    const stream = await this.stream(agentId, streamReq);
+    const text = await stream.text();
+    const resolvedThreadId = stream.threadId ?? threadId;
+
+    // 2. If we got text, return immediately
+    if (text && text.trim()) {
+      return { response: text, threadId: resolvedThreadId, status: "completed" };
+    }
+
+    // 3. Fallback: poll thread until processed, then fetch messages
+    const start = Date.now();
+    while (Date.now() - start < timeoutMs) {
+      try {
+        const thread = (await this.client.get(`/v1/agent-threads/${resolvedThreadId}`)).data as Record<string, unknown> | null;
+        const status = thread?.status as string | undefined;
+
+        if (status === "processed" || status === "idle") {
+          const history = (await this.client.get(`/v1/agent-threads/${resolvedThreadId}/messages`)).data as {
+            messages?: Array<{ role: string; content: string }>;
+          };
+          const assistantMsgs = history.messages?.filter((m) => m.role === "assistant") ?? [];
+          const lastMsg = assistantMsgs.length > 0 ? assistantMsgs[assistantMsgs.length - 1].content : "";
+          return { response: lastMsg, threadId: resolvedThreadId, status: "completed" };
+        }
+
+        if (status === "failed") {
+          return { response: "", threadId: resolvedThreadId, status: "failed" };
+        }
+      } catch {
+        // Thread not ready yet
+      }
+      await new Promise((r) => setTimeout(r, pollIntervalMs));
+    }
+
+    return { response: "", threadId: resolvedThreadId, status: "timeout" };
+  }
+
   // ── Upload File (authenticated) ────────────────────────────────────
   async uploadFile(agentId: string, payload: unknown): Promise<unknown> {
     return (await this.client.post(`/v1/agents/${agentId}/upload-file`, { json: payload })).data;
diff --git a/packages/sdk/src/resources/index.ts b/packages/sdk/src/resources/index.ts
@@ -1,4 +1,4 @@
-export { AgentsResource } from "./agents.js";
+export { AgentsResource, type AgentRunResult, type AgentRunOptions } from "./agents.js";
 export { WorkflowsResource } from "./workflows.js";
 export { ConnectionsResource } from "./connections.js";
 export { NodeTypesResource } from "./node-types.js";
diff --git a/packages/sdk/src/streaming.ts b/packages/sdk/src/streaming.ts
@@ -122,6 +122,12 @@ export class AgentStream implements AsyncIterable<StreamPart> {
   private _textChunks: string[] = [];
   private _processingPromise: Promise<void> | null = null;
 
+  /** Thread ID extracted from the first `data` event (type: thread_info). */
+  threadId: string | null = null;
+
+  /** User message ID extracted from the first `data` event. */
+  userMessageId: string | null = null;
+
   constructor(response: Response) {
     this.response = response;
   }
@@ -160,6 +166,27 @@ export class AgentStream implements AsyncIterable<StreamPart> {
     }
   }
 
+  /** Extract thread_id and user_message_id from data events. */
+  private _extractMeta(part: StreamPart): void {
+    if (part.type === "data" && Array.isArray(part.value)) {
+      for (const item of part.value) {
+        if (
+          item &&
+          typeof item === "object" &&
+          (item as Record<string, unknown>).type === "thread_info"
+        ) {
+          const data = (item as Record<string, unknown>).data as Record<string, unknown> | undefined;
+          if (data?.thread_id && !this.threadId) {
+            this.threadId = data.thread_id as string;
+          }
+          if (data?.user_message_id && !this.userMessageId) {
+            this.userMessageId = data.user_message_id as string;
+          }
+        }
+      }
+    }
+  }
+
   /**
    * Process the stream, dispatching events to listeners.
    * Automatically called by text(), parts(), or async iteration.
@@ -201,6 +228,7 @@ export class AgentStream implements AsyncIterable<StreamPart> {
           if (!part) continue;
 
           this._parts.push(part);
+          this._extractMeta(part);
 
           if (part.type === "textDelta") {
             this._textChunks.push(part.value as string);
@@ -215,6 +243,7 @@ export class AgentStream implements AsyncIterable<StreamPart> {
         const part = parseStreamLine(buffer);
         if (part) {
           this._parts.push(part);
+          this._extractMeta(part);
           if (part.type === "textDelta") {
             this._textChunks.push(part.value as string);
           }
@@ -272,6 +301,7 @@ export class AgentStream implements AsyncIterable<StreamPart> {
           if (!part) continue;
 
           this._parts.push(part);
+          this._extractMeta(part);
           if (part.type === "textDelta") {
             this._textChunks.push(part.value as string);
           }
@@ -285,6 +315,7 @@ export class AgentStream implements AsyncIterable<StreamPart> {
         const part = parseStreamLine(buffer);
         if (part) {
           this._parts.push(part);
+          this._extractMeta(part);
           if (part.type === "textDelta") {
             this._textChunks.push(part.value as string);
           }

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-export { AgentsResource } from "./agents.js";`
	`1`	`+export { AgentsResource, type AgentRunResult, type AgentRunOptions } from "./agents.js";`
`2`	`2`	`export { WorkflowsResource } from "./workflows.js";`
`3`	`3`	`export { ConnectionsResource } from "./connections.js";`
`4`	`4`	`export { NodeTypesResource } from "./node-types.js";`