> This page is for Alfa (Preview). > For clean Markdown of any page, append .md to the page URL. > For a complete documentation index, see https://docs.alfa.boosted.ai/llms.txt. > For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.alfa.boosted.ai/_mcp/server. # Streaming ## Overview The Thread API provides two methods for streaming real-time thread events: **gRPC bidirectional streaming** and **REST API Server-Sent Events (SSE)**. Both methods deliver incremental message chunks, artifacts, and completion events as they occur during thread execution. Choose the method that best fits your infrastructure: * **gRPC streaming** * **SSE streaming** ## Server-Sent Events (SSE) The SSE endpoint (`GET /v2/threads/stream/{thread_id}`) establishes a persistent HTTP connection that streams events as they occur. Open the stream connection **before** calling `POST /v2/threads/{thread_id}/messages` to capture all events. ### How SSE streaming works The server pushes events in this typical flow: 1. **`message_work_in_progress`** logs may appear before or between chunks, indicating background processing. 2. **`message_chunk`** events arrive incrementally as the AI generates its response. Each chunk contains the partial text under `message_chunk.message` and a `final_chunk` boolean flag. 3. When `final_chunk: true` arrives, the complete message is ready. 4. **`message`** events signal that a complete message has been assembled. 5. **`task_status`** or **`execution_status`** events report progress. When `status: "Complete"` appears, the thread is idle. ### Basic usage **`stream-thread.py`** ```python title="stream-thread.py" import os import requests BASE_URL = os.getenv("ALFA_API_BASE_URL", "https://sandbox.api.boosted.ai") ACCESS_TOKEN = os.environ["ALFA_ACCESS_TOKEN"] def auth_headers(): return {"Authorization": f"Bearer {ACCESS_TOKEN}"} def send_message(thread_id: str, message: str) -> dict: resp = requests.post( f"{BASE_URL}/v2/threads/{thread_id}/messages", headers=auth_headers(), data={"message": message}, timeout=30, ) resp.raise_for_status() return resp.json() ``` **`stream-thread.js`** ```javascript title="stream-thread.js" import EventSource from "eventsource"; const BASE_URL = process.env.ALFA_API_BASE_URL ?? "https://sandbox.api.boosted.ai"; const ACCESS_TOKEN = process.env.ALFA_ACCESS_TOKEN; function streamThread(threadId) { const url = `${BASE_URL}/v2/threads/stream/${threadId}`; const stream = new EventSource(url, { headers: { "Authorization": `Bearer ${ACCESS_TOKEN}` }, withCredentials: true, }); stream.onmessage = (event) => { const payload = JSON.parse(event.data); console.log(`[${payload.type}]`, payload.data); }; stream.onerror = (err) => { console.error("Stream closed", err); stream.close(); }; } ``` > **Info** > > If the SSE channel closes before completion, call `GET /v2/threads/{thread_id}/progress` to check status. Issue `DELETE /v2/threads/{thread_id}/task` when a user manually stops work. ### Complete SSE example The following example demonstrates the full workflow: creating a thread, establishing an SSE stream, sending a message, and processing real-time events until completion. The script uses Python's threading module to establish the SSE stream before sending the message, ensuring all events are captured: **`complete_sse_example.py`** ```python title="complete_sse_example.py" 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() ``` > **Note** > > Set the `ALFA_ACCESS_TOKEN` environment variable with your OAuth2 access token obtained via token exchange. See the [authorization guide](/alfa-preview/authorization) for details. The threading synchronization ensures the SSE listener is ready before the message triggers any events, preventing lost chunks. ### SSE event types The stream emits several event types. Here are the most common: **Work in progress logs** (background processing updates): ```json { "event": { "event_type": "message_work_in_progress", "log": "Analyzing financial statements..." } } ``` **Message chunks** (incremental response text): ```json { "event": { "event_type": "message_chunk", "message_chunk": { "message": "Tesla reported revenue of ", "message_id": "msg_xyz789" }, "final_chunk": false } } ``` **Final chunk** (last piece of the message): ```json { "event": { "event_type": "message_chunk", "message_chunk": { "message": "$25.2B in Q3 2024.", "message_id": "msg_xyz789" }, "final_chunk": true } } ``` **Complete message** (with metadata and citations): ```json { "event": { "event_type": "message", "message": { "message_id": "msg_xyz789", "message": "Apple reported revenue of $124.3B in Q1 2025, beating analyst expectations.", "message_metadata": { "tools_called": {}, "citations": [ { "citation_id": "cit_abc123", "citation_type": "web", "name": "Apple Q1 2025 Revenue Report", "summary": "Apple exceeded analyst expectations with record Q1 revenue.", "last_updated_at": "2026-02-19T10:00:00Z", "link": "https://example.com/aapl-q1-results", "last_fetched_at": "2026-02-19T09:30:00Z", "snippet_highlight_start": 120, "snippet_highlight_end": 350, "source_type": "web" } ] } } } } ``` Artifacts, when present, are JSON strings embedded in the message text (wrapped in triple backticks). Use `GET /v2/threads/{thread_id}/artifacts/{artifact_id}` to fetch the actual artifact data. **Task status** (progress updates): ```json { "event": { "event_type": "task_status", "status": "Complete", "log": "Analysis complete" } } ``` The message text is located at `event.message_chunk.message`. When `final_chunk: true` appears, the complete message is ready. The client must parse any artifact markers from the message text to extract `artifact_id` values. The script continues listening until it receives a `task_status` event with `status: "Complete"`, signaling that all processing has finished. ### SSE citation types Complete messages include a `citations` array in `message_metadata`. Each citation has a `citation_type` field that determines which additional fields are present. All citations share these common fields: | Field | Type | Description | | ----------------- | ------------------ | ---------------------------------------------- | | `citation_id` | `string` | Unique identifier for the citation | | `citation_type` | `string` | One of: `web`, `news_development`, `text` | | `name` | `string` | Display name of the source | | `summary` | `string \| null` | Brief summary of the cited content | | `last_updated_at` | `datetime \| null` | When the source was last updated | | `source_category` | `string \| null` | Category of the source | | `inline_offset` | `integer \| null` | Position offset in the message text | | `is_snippet` | `boolean` | Whether the citation references a text snippet | **Web citation** (`citation_type: "web"`): References a web page with a direct link and snippet highlight positions. ```json { "citation_id": "cit_001", "citation_type": "web", "name": "Apple Q1 2025 Revenue Report", "summary": "Apple exceeded analyst expectations with record Q1 revenue.", "last_updated_at": "2026-02-19T10:00:00Z", "link": "https://example.com/aapl-q1-results", "last_fetched_at": "2026-02-19T09:30:00Z", "snippet_highlight_start": 120, "snippet_highlight_end": 350, "source_type": "web" } ``` **News development citation** (`citation_type: "news_development"`): References a news development that aggregates multiple related articles. ```json { "citation_id": "cit_002", "citation_type": "news_development", "name": "Apple Announces New MacBook Pro Lineup", "summary": "Coverage of Apple's latest product announcement spanning 15 articles.", "last_updated_at": "2026-02-18T14:00:00Z", "num_articles": 15, "development_start": "2026-02-15T08:00:00Z" } ``` **Document citation** (`citation_type: "text"`): References a document such as an SEC filing, with snippet highlight positions. ```json { "citation_id": "cit_003", "citation_type": "text", "name": "AAPL 10-K Filing 2025", "summary": "Annual report filing for Apple Inc.", "last_updated_at": "2026-01-30T12:00:00Z", "snippet_highlight_start": 500, "snippet_highlight_end": 780, "is_snippet": true } ``` ## gRPC bidirectional streaming ### Protocol buffer definition Download and compile the following protocol buffer file to generate client stubs: **`thread_service.proto`** ```protobuf title="thread_service.proto" 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; } ``` ### Building the client Generate client code from the proto file using the gRPC tools for your language. **Python:** ```bash python -m grpc_tools.protoc \ -I./proto \ --python_out=. \ --grpc_python_out=. \ --pyi_out=. \ ``` This generates: * `thread_service_pb2.py`: Protocol buffer message classes * `thread_service_pb2_grpc.py`: gRPC service stub classes * `thread_service_pb2.pyi`: Type stubs for type checking **Go:** ```bash protoc \ -I./proto \ --go_out=. \ --go_opt=paths=source_relative \ --go-grpc_out=. \ --go-grpc_opt=paths=source_relative \ ``` This generates Go code locally. Update the `go_package` option in the proto file to match your module path, or import the generated code using the path specified in `go_package`. ### gRPC connection setup Establish a secure gRPC channel with keepalive settings to maintain the connection: **`grpc_setup.py`** ```python title="grpc_setup.py" import grpc GRPC_SERVER = "sandbox.api.boosted.ai:443" SERVICE_ACCOUNT_TOKEN = "" # Create secure channel with keepalive options credentials = grpc.ssl_channel_credentials() options = [ ('grpc.keepalive_time_ms', 10000), # Send keepalive ping every 10 seconds ('grpc.keepalive_timeout_ms', 5000), # Wait 5 seconds for ping ack ('grpc.keepalive_permit_without_calls', True), # Allow keepalive pings when no calls ('grpc.http2.max_pings_without_data', 0), # No limit on pings without data ('grpc.http2.min_time_between_pings_ms', 10000), # Min 10 seconds between pings ('grpc.http2.min_ping_interval_without_data_ms', 5000), # Min 5 seconds between pings without data ] channel = grpc.secure_channel(GRPC_SERVER, credentials, options=options) stub = thread_service_pb2_grpc.ThreadServiceStub(channel) metadata = [("authorization", f"Bearer {SERVICE_ACCOUNT_TOKEN}")] ``` **`grpc_setup.go`** ```go title="grpc_setup.go" package main import ( "context" "log" "time" "google.golang.org/grpc" "google.golang.org/grpc/credentials" "google.golang.org/grpc/keepalive" "google.golang.org/grpc/metadata" pb "" ) func main() { grpcServer := "sandbox.api.boosted.ai:443" serviceAccountToken := "" // Create connection with TLS and keepalive settings conn, err := grpc.NewClient( grpcServer, grpc.WithTransportCredentials(credentials.NewTLS(nil)), grpc.WithKeepaliveParams(keepalive.ClientParameters{ Time: 10 * time.Second, Timeout: 5 * time.Second, PermitWithoutStream: true, }), ) if err != nil { log.Fatalf("Failed to connect: %v", err) } defer conn.Close() // Create client client := pb.NewThreadServiceClient(conn) // Create context with authorization metadata ctx := context.Background() md := metadata.New(map[string]string{ "authorization": "Bearer " + serviceAccountToken, }) ctx = metadata.NewOutgoingContext(ctx, md) // Use ctx and client for RPC calls _ = ctx _ = client } ``` The keepalive settings ensure the connection remains active even during periods of inactivity, preventing timeouts. ### Complete gRPC example The following example demonstrates subscribing to thread events via gRPC streaming: **`grpc_streaming_example.py`** ```python title="grpc_streaming_example.py" 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 = "" THREAD_ID = "" 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() ``` ### How gRPC streaming works 1. **Create a secure channel** with SSL credentials and keepalive options to maintain the connection. 2. **Create a request generator** that yields the initial subscription request with `thread_id`, `user_token`, and `subscribe=True`. The generator continues running to keep the stream alive. 3. **Call the streaming RPC** using the generated stub, passing the request generator and authentication metadata. 4. **Iterate over responses** as they arrive. Each `StreamThreadEventsResponse` contains: * `thread_id`: The thread the event belongs to * `user_ids`: List of user IDs that requested access * `thread_data`: A `ThreadData` message containing the event payload The `thread_data` field is a `ThreadData` message with a `oneof payload` that can be one of: * `message_event`: A complete message with citations * `message_chunk_event`: An incremental message chunk with optional citations (check `final_chunk` for completion) * `work_in_progress_event`: Background processing status * `thread_name_event`: Thread name updates * `worker_error_event`: Error notifications * `message_cancelled_event`: Cancellation confirmations > **Tip** > > The request generator must continue yielding or sleeping to keep the bidirectional stream alive. If the generator exits, the stream will close. ### gRPC citation types Complete messages and message chunks both include a `citations` array. Each citation carries common metadata plus a type-specific object determined by `citation_type`: | Field | Type | Description | | ----------------- | ----------- | ---------------------------------------------------------------------------------------------------- | | `citation_id` | `UUID` | Unique identifier for the citation | | `name` | `string` | Display name of the source | | `citation_type` | `string` | One of: `web`, `news_development`, `news_article`, `document`, `fiscal_quarter_earnings`, `variable` | | `last_updated_at` | `Timestamp` | When the source was last updated | | `summary` | `string` | Brief summary of the cited content | **Web citation** (`web_citation`): References a web page with a direct link and snippet highlight positions. ```json { "citationId": {"id": "cit-001"}, "name": "Apple Q1 2025 Revenue Report", "citationType": "web", "lastUpdatedAt": "2026-02-19T10:00:00Z", "summary": "Apple exceeded analyst expectations with record Q1 revenue.", "webCitation": { "link": "https://example.com/aapl-q1-results", "lastFetchedAt": "2026-02-19T09:30:00Z", "snippetHighlightStart": 120, "snippetHighlightEnd": 350, "sourceType": "web" } } ``` **News development citation** (`news_development_citation`): References a news development that aggregates multiple related articles. ```json { "citationId": {"id": "cit-002"}, "name": "Apple Announces New MacBook Pro Lineup", "citationType": "news_development", "lastUpdatedAt": "2026-02-18T14:00:00Z", "summary": "Coverage of Apple's latest product announcement spanning 15 articles.", "newsDevelopmentCitation": { "numArticles": 15, "developmentStart": "2026-02-15T08:00:00Z" } } ``` **News article citation** (`news_article_citation`): References a specific news article with a direct link and article identifier. ```json { "citationId": {"id": "cit-003"}, "name": "Apple Q1 Results Beat Expectations", "citationType": "news_article", "lastUpdatedAt": "2026-02-17T16:30:00Z", "summary": "Apple exceeded analyst expectations for Q1 2026 earnings.", "newsArticleCitation": { "link": "https://example.com/aapl-q1-beat", "articleId": {"id": "art-456"} } } ``` **Document citation** (`document_citation`): References a document such as an SEC filing, with snippet highlight positions. ```json { "citationId": {"id": "cit-004"}, "name": "AAPL 10-K Filing 2025", "citationType": "document", "lastUpdatedAt": "2026-01-30T12:00:00Z", "summary": "Annual report filing for Apple Inc.", "documentCitation": { "snippetHighlightStart": 500, "snippetHighlightEnd": 780, "isSnippet": true } } ``` **Fiscal quarter earnings citation** (`fiscal_quarter_earnings_citation`): References an earnings call transcript or report for a specific fiscal quarter. ```json { "citationId": {"id": "cit-005"}, "name": "AAPL Q4 2025 Earnings Call", "citationType": "fiscal_quarter_earnings", "lastUpdatedAt": "2026-01-28T18:00:00Z", "summary": "Apple Q4 2025 earnings call transcript and financial report.", "fiscalQuarterEarningsCitation": { "snippetHighlightStart": 200, "snippetHighlightEnd": 450, "isSnippet": true, "callDatetime": "2025-10-30T17:00:00Z", "reportDatetime": "2025-10-30T16:30:00Z", "fiscalYear": 2025, "fiscalQuarter": 4 } } ``` **Variable citation** (`variable_citation`): References a computed financial variable or metric, including its formula, data provider, time range, and periodicity. ```json { "citationId": {"id": "cit-006"}, "name": "AAPL Revenue Growth", "citationType": "variable", "lastUpdatedAt": "2026-02-19T08:00:00Z", "summary": "Year-over-year revenue growth for Apple Inc.", "variableCitation": { "description": "Revenue growth rate", "dataProvider": "financial_data", "asOfTimestamp": "2026-02-19T00:00:00Z", "dateRangeStart": "2025-02-19T00:00:00Z", "dateRangeEnd": "2026-02-19T00:00:00Z", "periodStart": "Q1 2025", "periodEnd": "Q4 2025", "formula": "revenue_growth(AAPL)", "functionsUsed": [ {"name": "revenue_growth", "description": "Calculates YoY revenue growth"} ], "componentsUsed": [ {"name": "Revenue", "description": "Total revenue", "varId": "var-rev-001"} ], "variablePeriodicity": "VARIABLE_PERIODICITY_QUARTERLY" } } ``` ### Authentication gRPC streaming uses Bearer token authentication via metadata. Include your service account token in the metadata: ```python SERVICE_ACCOUNT_TOKEN = "" metadata = [("authorization", f"Bearer {SERVICE_ACCOUNT_TOKEN}")] stream = stub.StreamThreadEvents(request_generator(), metadata=metadata) ``` Obtain your service account token from your Boosted account. See the [authorization guide](/alfa-preview/authorization) for details. Ensure your token has permission to access the specified thread.