Skip to content
SafetyKit
StatusSupportDashboard
API Reference
Streams

Add parts

streams.add_parts(strnamespace, StreamAddPartsParams**kwargs) -> StreamAddPartsResponse
POST/v1/streams/{namespace}/parts

Append a list of parts to a stream by the stream identifier. The stream does not need to be created beforehand. This method returns immediately; parts are processed asynchronously.

ParametersExpand Collapse
namespace: str

The namespace to ingest stream data into

id: str
data: Iterable[Data]
One of the following:
class DataAudioURL: …
timestamp: str
type: Literal["audio_url"]
url: str
class DataFrameImageURL: …
timestamp: str
type: Literal["frame_image_url"]
url: str
class DataChatMessage: …
sender_display_name: str
sender_id: str
text: str
timestamp: str
type: Literal["chat_message"]
class DataTranscript: …
text: str
timestamp: str
type: Literal["transcript"]
speaker: Optional[str]
ReturnsExpand Collapse
class StreamAddPartsResponse: …

Stream ingestion request accepted for asynchronous processing.

request_id: str
status: Literal["accepted"]

Add parts

import os
from safetykit import Safetykit

client = Safetykit(
    api_key=os.environ.get("SAFETYKIT_API_KEY"),  # This is the default and can be omitted
)
response = client.streams.add_parts(
    namespace="namespace",
    id="YOUR_STREAM_ID",
    data=[{
        "type": "audio_url",
        "timestamp": "2026-03-10T18:54:20+00:00",
        "url": "https://example.com/234039.mp3",
    }],
)
print(response.request_id)
{
  "requestId": "req_01h2m7qdmdjckc30e1mnq6xqfd",
  "status": "accepted"
}
Returns Examples
{
  "requestId": "req_01h2m7qdmdjckc30e1mnq6xqfd",
  "status": "accepted"
}