Recording features are available in the Python SDK.
After a stream session ends, you can retrieve recording artifacts including the full video, events log, thumbnail, and preview.
Capturing the Stream ID
The stream ID is provided in the on_stream_started callback or returned from start_stream(). Save this ID to retrieve recordings later:
import asyncio
from odyssey import Odyssey
client = Odyssey( api_key = "ody_your_api_key_here" )
current_stream_id = None
def on_stream_started ( stream_id : str ) -> None :
global current_stream_id
current_stream_id = stream_id
print ( f "Stream started: { stream_id } " )
async def main ():
await client.connect(
on_video_frame = lambda frame : None ,
on_stream_started = on_stream_started,
)
# Or capture from return value
stream_id = await client.start_stream( "A cat" )
# ... interact with the stream ...
await client.end_stream()
await client.disconnect()
asyncio.run(main())
Retrieving a Recording
Use get_recording() with the stream ID to get presigned URLs for the recording artifacts:
recording = await client.get_recording(current_stream_id)
if recording.video_url:
print ( f "Video URL: { recording.video_url } " )
print ( f "Duration: { recording.duration_seconds } s" )
print ( f "Frames: { recording.frame_count } " )
if recording.events_url:
# Download and parse the events log (JSONL format)
import httpx
import json
async with httpx.AsyncClient() as http:
response = await http.get(recording.events_url)
events = [json.loads(line) for line in response.text.strip().split( ' \n ' )]
print ( f "Session events: { events } " )
get_recording() can be called without an active connection. It only requires a valid API key.
Recording Properties
Property Type Description stream_idstrThe stream ID video_urlstr | NonePresigned URL for full recording (MP4) events_urlstr | NonePresigned URL for events log (JSONL) thumbnail_urlstr | NonePresigned URL for thumbnail (JPEG) preview_urlstr | NonePresigned URL for preview video (MP4) frame_countint | NoneTotal frames in recording duration_secondsfloat | NoneRecording duration in seconds
Listing All Recordings
Use list_stream_recordings() to get a paginated list of all your recordings:
# Get recent recordings
result = await client.list_stream_recordings( limit = 10 )
print ( f "Found { result.total } recordings" )
for rec in result.recordings:
print ( f "Stream { rec.stream_id } : { rec.duration_seconds } s at { rec.width } x { rec.height } " )
# Paginate through results
page2 = await client.list_stream_recordings( limit = 10 , offset = 10 )
Option Type Default Description limitint | NoneNoneMaximum recordings to return offsetint | NoneNoneNumber of recordings to skip
Complete Workflow Example
This example shows the full workflow: starting a stream, interacting with it, ending it, and then retrieving the recording:
import asyncio
from odyssey import Odyssey, OdysseyConnectionError
async def main ():
client = Odyssey( api_key = "ody_your_api_key_here" )
try :
await client.connect(
on_video_frame = lambda frame : print ( f "Frame: { frame.width } x { frame.height } " ),
)
# Start an interactive stream and capture the stream ID
stream_id = await client.start_stream( "A cat sitting on a windowsill" , portrait = True )
print ( f "Stream started: { stream_id } " )
# Send some interactions
await client.interact( "The cat stretches and yawns" )
await client.interact( "The cat looks outside at birds" )
# Let it run for a bit
await asyncio.sleep( 5 )
# End the stream
await client.end_stream()
except OdysseyConnectionError as e:
print ( f "Connection failed: { e } " )
finally :
await client.disconnect()
# Retrieve the recording (can be done after disconnect)
recording = await client.get_recording(stream_id)
if recording.video_url:
print ( f "Recording available at: { recording.video_url } " )
print ( f "Duration: { recording.duration_seconds } seconds" )
asyncio.run(main())
Downloading Recordings
Example of downloading recording files:
import httpx
from pathlib import Path
async def download_recording ( client : Odyssey, stream_id : str , output_dir : Path):
recording = await client.get_recording(stream_id)
async with httpx.AsyncClient() as http:
if recording.video_url:
response = await http.get(recording.video_url)
(output_dir / f " { stream_id } .mp4" ).write_bytes(response.content)
print ( f "Downloaded video: { stream_id } .mp4" )
if recording.thumbnail_url:
response = await http.get(recording.thumbnail_url)
(output_dir / f " { stream_id } _thumb.jpg" ).write_bytes(response.content)
print ( f "Downloaded thumbnail: { stream_id } _thumb.jpg" )
Broadcast & Recordings
Broadcast streams are recorded the same way as standard interactive streams.
Recordings are tied to the stream ID , not to individual spectators.
Starting a stream with broadcast=True does not change how recordings are created.
Spectators do not generate separate recordings.
Use the stream_id returned from start_stream() (or received via on_stream_started) to retrieve artifacts with get_recording().
Recording Types Python dataclasses for recordings
Odyssey Client get_recording and list_stream_recordings methods