Streaming

import json
import os
import sys
import threading
import time
import requests
BASE_URL = "https://sandbox.api.boosted.ai"
ACCESS_TOKEN = os.environ.get("ALFA_ACCESS_TOKEN", "your-access-token-here")
stream_connected = threading.Event()
stream_error = threading.Event()
def create_thread(thread_name):
    url = f"{BASE_URL}/v2/threads"
    headers = {
        "Authorization": f"Bearer {ACCESS_TOKEN}",
        "Content-Type": "application/json"
    }
    data = {"thread_name": thread_name}
    
    response = requests.post(url, headers=headers, json=data)
    response.raise_for_status()
    return response.json()["thread_id"]
def send_message(thread_id, message):
    url = f"{BASE_URL}/v2/threads/{thread_id}/messages"
    headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"}
    data = {"message": message}
    
    print(f"\nSending message...")
    response = requests.post(url, headers=headers, data=data)
    response.raise_for_status()
    print("Message sent successfully!")
    return response.json()
def stream_thread_events(thread_id):
    url = f"{BASE_URL}/v2/threads/stream/{thread_id}"
    headers = {
        "Authorization": f"Bearer {ACCESS_TOKEN}",
        "Accept": "text/event-stream",
        "Cache-Control": "no-cache"
    }
    
    try:
        print("\nConnecting to event stream...")
        response = requests.get(url, headers=headers, stream=True, timeout=None)
        response.raise_for_status()
        
        print("Connected to stream, waiting for events...\n")
        stream_connected.set()
        
        for line in response.iter_lines(decode_unicode=True):
            if line:
                if line.startswith(':'):
                    continue
                    
                if line.startswith('data: '):
                    data = line[6:]
                    try:
                        event_data = json.loads(data)
                        event_type = event_data.get('event', {}).get('event_type', 'unknown')
                        
                        if event_type == 'message_chunk':
                            chunk = event_data['event'].get('message_chunk', {})
                            sys.stdout.write(chunk.get('message', ''))
                            sys.stdout.flush()
                            
                            if event_data['event'].get('final_chunk'):
                                print("\n-- final chunk --")
                        elif event_type == 'message':
                            print(f"\n\nComplete message received")
                        elif event_type == 'task_status' or event_type == 'execution_status':
                            status = event_data['event'].get('status', '')
                            if status == 'Complete':
                                print(f"\n\nTask completed!")
                                break
                        else:
                            print(f"\n[{event_type}]")
                    except json.JSONDecodeError:
                        print(f"Data: {data}")
                elif line.startswith('event: '):
                    event_type = line[7:]
                    if event_type == 'close':
                        print("\n\nStream closed by server")
                        break
                    
    except requests.exceptions.RequestException as e:
        print(f"\nStream error: {e}")
        stream_error.set()
    except Exception as e:
        print(f"\nUnexpected error: {e}")
        stream_error.set()
def main():
    try:
        print("Creating a new thread...")
        thread_id = create_thread("Research Thread")
        print(f"Thread created with ID: {thread_id}")
        
        message = input("\nEnter your message: ")
        
        stream_thread = threading.Thread(
            target=stream_thread_events, 
            args=(thread_id,), 
            daemon=True
        )
        stream_thread.start()
        
        if not stream_connected.wait(timeout=10):
            print("Failed to connect to stream within 10 seconds")
            return
        
        if stream_error.is_set():
            print("Stream connection failed")
            return
        
        time.sleep(0.5)
        
        send_message(thread_id, message)
        
        stream_thread.join(timeout=120)
        
        if stream_thread.is_alive():
            print("\n\nStream still active after 120 seconds, exiting...")
        
    except requests.exceptions.RequestException as e:
        print(f"Error: {e}")
        sys.exit(1)
    except KeyboardInterrupt:
        print("\n\nInterrupted by user")
        sys.exit(0)
if __name__ == "__main__":
    if ACCESS_TOKEN == "your-access-token-here":
        print("Please set your ALFA_ACCESS_TOKEN environment variable before running.")
        sys.exit(1)
    main()

syntax = "proto3";
package thread_service.v1;
import "google/protobuf/wrappers.proto";
import "google/protobuf/timestamp.proto";
option java_package = "com.gbi.protobuf_defs.generated.thread_service.v1";
option go_package = "github.com/GBI-Core/protobuf-defs/gobuild/thread_service/v1";
service ThreadService {
  /////////////////////////////////////////////////////////////////
  // Thread API
  /////////////////////////////////////////////////////////////////
  // Streams thread events from multiple threads
  rpc StreamThreadEvents(stream StreamThreadEventsRequest)
      returns (stream StreamThreadEventsResponse);
}
/////////////////////////////////////////////////////////////////
// Common messages
/////////////////////////////////////////////////////////////////
message UUID {
  string id = 1;
}
message Citation {
  message WebCitation {
    string link = 1;
    google.protobuf.Timestamp last_fetched_at = 2;
    int64 snippet_highlight_start = 3;
    int64 snippet_highlight_end = 4;
    string source_type = 5;
  }
  message NewsDevelopmentCitation {
    int64 num_articles = 1;
    google.protobuf.Timestamp development_start = 2;
  }
  message NewsArticleCitation {
    string link = 1;
    UUID article_id = 2;
  }
  message DocumentCitation {
    int64 snippet_highlight_start = 1;
    int64 snippet_highlight_end = 2;
    bool is_snippet = 3;
  }
  message FiscalQuarterEarningsCitation {
    int64 snippet_highlight_start = 1;
    int64 snippet_highlight_end = 2;
    bool is_snippet = 3;
    google.protobuf.Timestamp call_datetime = 4;
    google.protobuf.Timestamp report_datetime = 5;
    google.protobuf.Int32Value fiscal_year = 6;
    google.protobuf.Int32Value fiscal_quarter = 7;
  }
  message VariableCitation {
    enum VariablePeriodicity {
      VARIABLE_PERIODICITY_UNSPECIFIED = 0;
      VARIABLE_PERIODICITY_DAILY = 1;
      VARIABLE_PERIODICITY_MONTHLY = 2;
      VARIABLE_PERIODICITY_QUARTERLY = 3;
      VARIABLE_PERIODICITY_YEARLY = 4;
    }
    message FunctionMetadata {
      string name = 1;
      string description = 2;
    }
    message VariableMetadata {
      string name = 1;
      string description = 2;
      string var_id = 3;
    }
    string description = 1;
    string data_provider = 2;
    google.protobuf.Timestamp as_of_timestamp = 3;
    google.protobuf.Timestamp date_range_start = 4;
    google.protobuf.Timestamp date_range_end = 5;
    string period_start = 6;
    string period_end = 7;
    string formula = 8;
    repeated FunctionMetadata functions_used = 9;
    repeated VariableMetadata components_used = 10;
    VariablePeriodicity variable_periodicity = 11;
  }
  UUID citation_id = 1;
  string name = 2;
  string citation_type = 3;
  google.protobuf.Timestamp last_updated_at = 4;
  string summary = 8;
  oneof citation {
    WebCitation web_citation = 5;
    NewsDevelopmentCitation news_development_citation = 6;
    DocumentCitation document_citation = 7;
    NewsArticleCitation news_article_citation = 9;
    FiscalQuarterEarningsCitation fiscal_quarter_earnings_citation = 10;
    VariableCitation variable_citation = 11;
  }
}
message ThreadData {
  enum ErrorType {
    ERROR_TYPE_UNKNOWN = 0;
    ERROR_TYPE_INTERNAL = 1;
  }
  message MessageEvent {
    UUID message_id  = 1;
    string message = 2;
    bool is_user_message = 3;
    google.protobuf.Timestamp message_time = 4;
    repeated Citation citations = 5;
  }
  message MessageChunkEvent {
    UUID message_id  = 1;
    string message = 2;
    bool final_chunk = 3;
    repeated Citation citations = 4;
  }
  message WorkInProgressEvent {
    string log = 1;
  }
  message ThreadNameEvent {
    string thread_name = 1;
  }
  message WorkerErrorEvent {
    ErrorType error_type = 1;
  }
  message MessageCancelledEvent {
    string cancellation_message = 1;
  }
  oneof payload {
    MessageEvent message_event = 1;
    MessageChunkEvent message_chunk_event = 2;
    WorkInProgressEvent work_in_progress_event = 3;
    ThreadNameEvent thread_name_event = 4;
    WorkerErrorEvent worker_error_event = 5;
    MessageCancelledEvent message_cancelled_event = 6;
  }
}
/////////////////////////////////////////////////////////////////
// REQUEST/RESPONSE MESSAGES
/////////////////////////////////////////////////////////////////
message StreamThreadEventsRequest {
  // The token for the user that is requesting access to the thread
  string user_token = 1;
  // The thread id the user is requesting access to
  UUID thread_id = 2;
  bool subscribe = 3;
}
message StreamThreadEventsResponse {
  // The user(s) that requested access to the thread
  repeated UUID user_ids = 1;
  // The thread id the data corresponds to
  UUID thread_id = 2;
  // The data for the thread
  ThreadData thread_data = 3;
}

import time
from collections.abc import Iterator
import grpc
from thread_service.v1 import thread_service_pb2, thread_service_pb2_grpc
def test_streaming():
    # Configuration - update these
    GRPC_SERVER = "sandbox.api.boosted.ai:443"
    SERVICE_ACCOUNT_TOKEN = "<YOUR SERVICE ACCOUNT TOKEN>"
    THREAD_ID = "<YOUR THREAD ID>"
    USER_TOKEN = "<YOUR USER TOKEN>"
    # Connect
    credentials = grpc.ssl_channel_credentials()
    options = [
        ('grpc.keepalive_time_ms', 10000),
        ('grpc.keepalive_timeout_ms', 5000),
        ('grpc.keepalive_permit_without_calls', True),
        ('grpc.http2.max_pings_without_data', 0),
        ('grpc.http2.min_time_between_pings_ms', 10000),
        ('grpc.http2.min_ping_interval_without_data_ms', 5000),
    ]
    channel = grpc.secure_channel(GRPC_SERVER, credentials, options=options)
    stub = thread_service_pb2_grpc.ThreadServiceStub(channel)
    metadata = [("authorization", f"Bearer {SERVICE_ACCOUNT_TOKEN}")]
    print(f"Connecting to {GRPC_SERVER}")
    def request_generator():
        """Send subscription request."""
        yield thread_service_pb2.StreamThreadEventsRequest(
            thread_id=thread_service_pb2.UUID(id=THREAD_ID),
            user_token=USER_TOKEN,
            subscribe=True,
        )
        # Keep stream alive
        while True:
            time.sleep(60)
    try:
        # Start streaming
        stream: Iterator[thread_service_pb2.StreamThreadEventsResponse] = stub.StreamThreadEvents(
            request_generator(), metadata=metadata
        )
        print(f"Subscribed to thread {THREAD_ID}")
        print("Listening for events... (Ctrl+C to stop)")
        # Receive events
        for response in stream:
            print("\nEvent received!")
            print(f"   Thread: {response.thread_id.id}")
            print(f"   Users: {[u.id for u in response.user_ids]}")
            print(f"   Data: {response.thread_data}")
    except KeyboardInterrupt:
        print("\nStopped")
    except grpc.RpcError as e:
        print(f"Error: {e.code()}: {e.details()}")
    finally:
        channel.close()
if __name__ == "__main__":
    test_streaming()