Developer Quickstart

pip install archetypeai
export ATAI_API_KEY="insert_your_key"

import argparse
from archetypeai.api_client import ArchetypeAI

def main(args):
    # Create a new client using your unique API key.
    client = ArchetypeAI(args.api_key)

    # Upload the video file to the archetype platform.
    file_response = client.files.local.upload(args.filename)

    # Define a custom lens that reads from a video and writes to an SSE stream.
    client.lens.create_and_run_lens(f"""
       lens_name: Custom Activity Monitor
       lens_config:
        model_parameters:
            model_version: Newton::c2_3_7b_2508014e10af56
            instruction: Analyze the video with the following focus.
            focus: Delivery
            temporal_focus: 5
            max_new_tokens: 128
        input_streams:
            - stream_type: video_file_reader
              stream_config:
                file_id: {file_response['file_id']}
        output_streams:
            - stream_type: server_side_events_writer
    """, session_callback, client=client)

def session_callback(session_id: str, session_endpoint: str, client: ArchetypeAI):
    """Main function to run the logic of a custom lens session."""

    # Create a SSE reader to read the output of the lens.
    sse_reader = client.lens.sessions.create_sse_consumer(session_id)

    # Read events from the SSE stream until either the last message is received.
    for event in sse_reader.read(block=True):
        print(f"[sse_reader] {event}")

    # Close any active reader.
    sse_reader.close()

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--api_key", required=True, type=str)
    parser.add_argument("--filename", required=True, type=str)
    args = parser.parse_args()
    main(args)

python example.py --api_key=$ATAI_API_KEY --filename=your_test_video.mp4