> ## Documentation Index
> Fetch the complete documentation index at: https://docs.rkat.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Python SDK

> Get started with the Meerkat Python SDK: sessions, streaming, schedules, mobs, and typed runtime contracts.

The Python SDK (`meerkat-sdk`) is a runtime-backed client over `rkat-rpc`.

* **Contract version:** `0.6.6`
* **Python:** `>=3.10`
* **Runtime dependencies:** `websockets>=12,<16`

## Install

```bash theme={null}
pip install meerkat-sdk
```

For local development:

```bash theme={null}
pip install -e "sdks/python[dev]"
```

## Quick start

```python theme={null}
import asyncio
from meerkat import MeerkatClient, TextDelta

async def main() -> None:
    async with MeerkatClient() as client:
        session = await client.create_session("What is the capital of France?")
        print(session.text)

        result = await session.turn("And Germany?")
        print(result.text)

        async with session.stream("Give one extra fact.") as events:
            async for event in events:
                match event:
                    case TextDelta(delta=chunk):
                        print(chunk, end="", flush=True)
            print()

        await session.archive()

asyncio.run(main())
```

## Method overview

### Session lifecycle

* `await client.create_session(...) -> Session`
* `client.create_session_streaming(...) -> EventStream`
* `await client.create_deferred_session(...) -> DeferredSession`
* `await client.list_sessions(...) -> list[SessionSummary]`
* `await client.read_session(session_id) -> SessionDetails`
* `await client.read_session_history(session_id, offset=0, limit=None) -> SessionHistory`

### Session runtime inputs

* `await client.inject_context(session_id, text, source=None, idempotency_key=None)`
* `await session.inject_context(...)`
* `await client.send_external_event(session_id, event_type, payload, blocks=None)`
* `await session.send_external_event(event_type, payload, blocks=None)`

### Turn APIs

`Session.turn(...)`, `Session.stream(...)`, and `DeferredSession.start_turn(...)` support:

* `skill_refs`, `flow_tool_overlay`
* `additional_instructions`
* `keep_alive`, `model`, `provider`, `max_tokens`, `system_prompt`
* `output_schema`, `structured_output_retries`, `provider_params`

### Config APIs

* `await client.get_config() -> ConfigEnvelope`
* `await client.set_config(config, expected_generation=None) -> ConfigEnvelope`
* `await client.patch_config(patch, expected_generation=None) -> ConfigEnvelope`

All config APIs return a **config envelope** (`config`, `generation`, realm metadata), not a bare config object.

### Additional runtime surfaces

* `await client.get_models_catalog()`
* `await client.create_schedule(...)`
* `await client.get_schedule(...)`
* `await client.list_schedules(labels=None, limit=None, offset=None)`
* `await client.update_schedule(...)`
* `await client.pause_schedule(...)`
* `await client.resume_schedule(...)`
* `await client.delete_schedule(...)`
* `await client.list_schedule_occurrences(...)`
* `await client.list_schedule_tools()`
* `await client.call_schedule_tool(...)`

### Live channel APIs

The old `RealtimeChannel` helper has been removed. Use `LiveChannel.session(...)`
for a session-bound wrapper, or call the direct `live/*` RPC helpers on
`MeerkatClient`.

```python theme={null}
from meerkat import LiveChannel, MeerkatClient

client = MeerkatClient()
await client.connect(live_ws=True)
try:
    session = await client.create_session("Open a live channel")
    channel = LiveChannel.session(client, session.id, turning_mode="explicit_commit")

    opened = await channel.open()
    # Connect to opened["transport"]["url"] externally.
    await channel.send_input_text("hello")
    await channel.commit_input(response_modality="text")
    await channel.close()
finally:
    await client.close()
```

* `await client.live_open(session_id, turning_mode=None)`
* `await client.live_status(channel_id)`
* `await client.live_close(channel_id)`
* `await client.live_send_input_text(channel_id, text)`
* `await client.live_send_input_audio(channel_id, data_base64, sample_rate_hz, channels)`
* `await client.live_send_input_image(channel_id, mime, data_base64)`
* `await client.live_send_input_video_frame(channel_id, codec, data_base64, timestamp_ms)`
* `await client.live_commit_input(channel_id, response_modality=None)`
* `await client.live_interrupt(channel_id)`
* `await client.live_truncate(channel_id, item_id, content_index, audio_played_ms)`
* `await client.live_refresh(channel_id)`

### Mob APIs

* `await client.create_mob(...) -> Mob`
* `await client.list_mobs()`
* `await mob.spawn(...)`
* `await mob.spawn_many(...)`
* `await mob.read_events(after_cursor=0, limit=100)`
* `await mob.wait_for_kickoff_complete(...)`
* `await mob.spawn_helper(..., role_name=...)`
* `await mob.fork_helper(..., role_name=...)`

### Realm profile CRUD

* `await client.create_mob_profile(name, profile)`
* `await client.get_mob_profile(name) -> StoredMobProfile | None`
* `await client.list_mob_profiles()`
* `await client.update_mob_profile(name, profile, expected_revision=...)`
* `await client.delete_mob_profile(name, expected_revision=...)`

### Additional wrappers

The Python client also exposes:

* realm helpers: `list_realms()`, `get_realm(...)`
* auth helpers: `list_auth_profiles(...)`, `get_auth_profile(...)`, `create_auth_profile(...)`, `delete_auth_profile(...)`, `auth_login_*`, `auth_provision_api_key(...)`, `auth_status(...)`, `auth_logout(...)`
* blob/skill/MCP helpers: `get_blob(...)`, `list_skills()`, `mcp_add(...)`, `mcp_remove(...)`, `mcp_reload(...)`
* comms helpers: `session.send(...)`, `session.peers()`
* typed ingress helpers: `send_peer_response_terminal(...)`

### Generated assistant images

When a session model calls the built-in `generate_image` tool, generated images appear in committed history as `SessionAssistantBlock` entries with `block_type == "image"`. The SDK preserves `image_id`, `blob_id`, `media_type`, `width`, `height`, `revised_prompt`, and provider metadata.

```python theme={null}
history = await session.history()
image = next(
    block
    for message in history.messages
    for block in message.blocks
    if block.block_type == "image"
)
payload = await client.get_blob(image.blob_id)
```

Image-generation wire contracts are also exported for integrations that inspect tool results directly: `WireGenerateImageRequest`, `WireGenerateImageExecutionPlan`, `WireImageGenerationToolResult`, `WireImageOperationPhase`, and `WireAssistantImageRef`.

## Typed session metadata

* `SessionSummary` (from `session/list`) includes `total_tokens`
* `SessionDetails` (from `session/read`) includes `model`, `provider`, `last_assistant_text`
* both use integer unix timestamps for `created_at` / `updated_at`

## Streaming model

`EventStream` is an async context manager returned by `session.stream()` and `client.create_session_streaming()`. Iterate it to receive typed event dataclasses, then read `events.result`.

```python theme={null}
async with session.stream("Summarize logs") as events:
    async for event in events:
        ...
    result = events.result
```