Python Quickstart

import asyncio
from anam import AnamClient

async def main():
    client = AnamClient(
        api_key="your-api-key",
        persona_id="your-persona-id",
    )

    async with client.connect() as session:
        async def consume_video():
            async for frame in session.video_frames():
                img = frame.to_ndarray(format="rgb24")
                print(f"Video: {frame.width}x{frame.height}")

        async def consume_audio():
            async for frame in session.audio_frames():
                samples = frame.to_ndarray()
                print(f"Audio: {samples.size} samples")

        await asyncio.gather(consume_video(), consume_audio())

asyncio.run(main())

async with client.connect() as session:
    await session.talk("Hello, tell me about yourself.")

from anam import AnamClient, AnamEvent

client = AnamClient(
    api_key="your-api-key",
    persona_id="your-persona-id",
)

@client.on(AnamEvent.CONNECTION_ESTABLISHED)
async def on_connected():
    print("Connected to persona")

@client.on(AnamEvent.MESSAGE_STREAM_EVENT_RECEIVED)
async def on_message(event):
    print(f"{event.role}: {event.content}")

await client.connect()

from anam import AnamEvent, Message, MessageRole

@client.on(AnamEvent.MESSAGE_HISTORY_UPDATED)
async def on_message_history(messages: list[Message]):
    message = messages[-1]

    if message.role != MessageRole.ASSISTANT:
        return

    if message.utterances:
        for utterance in message.utterances:
            print(f"[{utterance.id}] {utterance.content}")
    else:
        print(message.content)