Skip to main content
The Python SDK (meerkat-sdk) lets you manage sessions, run agent turns, stream events, and query capabilities from Python. It communicates with a local rkat rpc subprocess over JSON-RPC 2.0.
  • Python: >=3.10
  • Dependencies: zero runtime dependencies

​
Getting started

1

Install the SDK

pip install sdks/python
2

Install the rkat binary

The SDK spawns rkat rpc as a subprocess. Build it from source:
cargo build -p meerkat-cli --release
# Binary is at target/release/rkat
3

Connect and run

import asyncio
from meerkat import MeerkatClient

async def main():
    client = MeerkatClient()
    await client.connect()

    result = await client.create_session("What is the capital of France?")
    print(result.text)
    print(result.session_id)

    # Multi-turn
    result2 = await client.start_turn(result.session_id, "And of Germany?")
    print(result2.text)

    await client.archive_session(result.session_id)
    await client.close()

asyncio.run(main())
If the rkat binary is not on PATH, pass its location explicitly:
client = MeerkatClient(rkat_path="/path/to/rkat")

​
Method overview

MethodDescription
connect()Spawn rkat rpc, handshake, fetch capabilities
close()Terminate the subprocess
create_session(prompt, ...)Create a session and run the first turn
start_turn(session_id, prompt)Continue an existing session
interrupt(session_id)Cancel an in-flight turn
list_sessions()List active sessions
read_session(session_id)Read session state
archive_session(session_id)Remove a session
get_capabilities()Get runtime capabilities
has_capability(id)Check if a capability is available
require_capability(id)Guard a code path by capability
get_config()Read runtime configuration
set_config(config)Replace runtime configuration
patch_config(patch)Merge-patch runtime configuration

​
MeerkatClient

​
Constructor

MeerkatClient(rkat_path: str = "rkat")
Raises MeerkatError (code "BINARY_NOT_FOUND") if the binary is not found.

​
connect()

async def connect(self) -> MeerkatClient
Starts the rkat rpc subprocess, performs the initialize handshake, checks contract version compatibility, and fetches capabilities. Returns self for chaining.

​
close()

async def close(self) -> None
Sends SIGTERM and waits up to 5 seconds; falls back to SIGKILL on timeout.

​
create_session()

async def create_session(
    self,
    prompt: str,
    model: Optional[str] = None,
    provider: Optional[str] = None,
    system_prompt: Optional[str] = None,
    max_tokens: Optional[int] = None,
    output_schema: Optional[dict] = None,
    structured_output_retries: int = 2,
    hooks_override: Optional[dict] = None,
    enable_builtins: bool = False,
    enable_shell: bool = False,
    enable_subagents: bool = False,
    enable_memory: bool = False,
    host_mode: bool = False,
    comms_name: Optional[str] = None,
    provider_params: Optional[dict] = None,
) -> WireRunResult
Creates a new session and runs the first turn. Returns a WireRunResult.

​
start_turn()

async def start_turn(self, session_id: str, prompt: str) -> WireRunResult
Starts a new turn on an existing session.

​
interrupt()

async def interrupt(self, session_id: str) -> None
Cancels a running turn.

​
Session management

sessions = await client.list_sessions()
info = await client.read_session(session_id)
await client.archive_session(session_id)

​
Config management

config = await client.get_config()
await client.set_config(config)
updated = await client.patch_config({"max_tokens": 2048})

​
Examples

async def multi_turn():
    client = MeerkatClient()
    await client.connect()

    result = await client.create_session(
        "You are a helpful math tutor. What is 12 * 15?",
        model="claude-sonnet-4-5",
    )
    print(f"Turn 1: {result.text}")

    result2 = await client.start_turn(result.session_id, "Now divide that result by 3.")
    print(f"Turn 2: {result2.text}")

    await client.archive_session(result.session_id)
    await client.close()

async def structured():
    client = MeerkatClient()
    await client.connect()

    schema = {
        "type": "object",
        "properties": {
            "city": {"type": "string"},
            "country": {"type": "string"},
            "population": {"type": "integer"},
        },
        "required": ["city", "country", "population"],
    }

    result = await client.create_session(
        "Give me information about Tokyo.",
        output_schema=schema,
        structured_output_retries=3,
    )
    print(f"Structured: {result.structured_output}")

    await client.archive_session(result.session_id)
    await client.close()

async def with_tools():
    client = MeerkatClient()
    await client.connect()

    result = await client.create_session(
        "Create a file called hello.txt with 'Hello World' in it",
        enable_builtins=True,
        enable_shell=True,
    )
    print(f"Response: {result.text}")
    print(f"Tool calls made: {result.tool_calls}")

    await client.archive_session(result.session_id)
    await client.close()

​
See also