The Complete Production SOP: How to Deploy and Run OpenAI Assistants API v2 Threads & Streaming in Autonomous AI Agents

---
name: "openai-assistants-v2-threads"
description: "Build stateful conversational AI applications with OpenAI Assistants API v2."
metadata:
  author: "OpenAI"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "create openai assistant thread"
  - "stream assistant run"
  - "attach file_search to assistant"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

# OpenAI Assistants API v2 Threads & Streaming

Implements OpenAI Assistants API: persistent Threads, Messages, Runs, tool attachments (file_search, code_interpreter), and real-time streaming with AssistantEventHandler.

## When to Invoke
Invoke this skill whenever you need to:
- create openai assistant thread
- stream assistant run
- attach file_search to assistant

## Execution SOP & Rules
1. Create persistent threads with client.beta.threads.create() to manage dialogue state automatically.
2. Use client.beta.threads.runs.stream() with custom AssistantEventHandler for instant token streaming.
3. Submit tool outputs via client.beta.threads.runs.submit_tool_outputs_stream() when actions are requested.
4. Poll run status or handle webhooks to detect run completion, failures, and cancellations.

import os
from typing import Any
from openai import OpenAI, AssistantEventHandler
from dotenv import load_dotenv

load_dotenv('.env.local')
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

class CustomEventHandler(AssistantEventHandler):
    def on_text_created(self, text) -> None:
        print("\nassistant > ", end="", flush=True)

    def on_text_delta(self, delta, snapshot):
        print(delta.value, end="", flush=True)

    def on_tool_call_created(self, tool_call):
        print(f"\nassistant [tool] > {tool_call.type}\n", flush=True)

def stream_conversation(thread_id: str, assistant_id: str, user_content: str) -> None:
    client.beta.threads.messages.create(
        thread_id=thread_id,
        role="user",
        content=user_content
    )
    handler = CustomEventHandler()
    with client.beta.threads.runs.stream(
        thread_id=thread_id,
        assistant_id=assistant_id,
        event_handler=handler
    ) as stream:
        stream.until_done()

import OpenAI from 'openai';
import 'dotenv/config';

const client = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

export async function createThreadAndStreamRun(
  vectorStoreId: string,
  assistantId: string,
  prompt: string
) {
  // 1. Create persistent thread with file_search resources attached
  const thread = await client.beta.threads.create({
    tool_resources: {
      file_search: {
        vector_store_ids: [vectorStoreId]
      }
    },
    messages: [
      { role: 'user', content: prompt }
    ]
  });

  // 2. Stream assistant run using the v2 event stream handler
  const stream = client.beta.threads.runs.stream(thread.id, {
    assistant_id: assistantId
  });

  stream.on('textDelta', (textDelta) => {
    process.stdout.write(textDelta.value ?? '');
  });

  await stream.finalRun();
  return thread.id;
}