Python SDK reference

Wire types

WireRunResult

from meerkat import WireRunResult

Returned by create_session() and start_turn().

session_id

str

Session UUID.

text

str

Agent response text.

turns

int

Number of turns completed.

tool_calls

int

Number of tool calls made.

usage

Optional[WireUsage]

Token usage breakdown.

structured_output

Optional[Any]

Parsed structured output.

schema_warnings

Optional[list]

Schema validation warnings.

WireUsage

from meerkat import WireUsage

input_tokens

int

Input tokens consumed.

output_tokens

int

Output tokens generated.

total_tokens

int

Total tokens.

cache_creation_tokens

Optional[int]

Cache creation tokens.

cache_read_tokens

Optional[int]

Cache read tokens.

WireEvent (deprecated)

from meerkat import WireEvent

session_id

str

Session this event belongs to.

event

Optional[dict]

The agent event payload.

WireEvent is deprecated. The streaming API (StreamingTurn) yields raw event dicts directly. This type is retained for backward compatibility only.

CapabilitiesResponse

from meerkat import CapabilitiesResponse, CapabilityEntry

contract_version

str

Server contract version.

capabilities

list[CapabilityEntry]

List of capability entries.

Each CapabilityEntry has id, description, and status fields.

CapabilityChecker

Standalone capability-checking helper:

from meerkat import CapabilityChecker

caps = await client.get_capabilities()
checker = CapabilityChecker(caps)

if checker.has("shell"):
    print("Shell tool is available")

checker.require("comms")  # raises CapabilityUnavailableError if unavailable

print(checker.available)  # e.g. ["sessions", "streaming", "builtins"]

Method	Description
`has(capability_id)`	Returns `True` if status is `"Available"`
`require(capability_id)`	Raises `CapabilityUnavailableError` if not available
`available`	List of all available capability IDs

SkillHelper

Helpers for invoking Meerkat skills:

from meerkat import SkillHelper

helper = SkillHelper(client)

if helper.is_available():
    result = await helper.invoke(session_id, "/shell-patterns", "How do I run a background job?")
    print(result.text)

    result = await helper.invoke_new_session("/code-review", "Review this function", model="claude-sonnet-4-5")

Method	Description
`is_available()`	Returns `True` if `"skills"` capability is available
`require_skills()`	Raises error if skills not available
`invoke(session_id, skill_ref, prompt)`	Invoke skill in existing session
`invoke_new_session(skill_ref, prompt, model?)`	Create session and invoke skill

Streaming API

StreamingTurn provides real-time event streaming during agent execution. Events are yielded as raw dicts matching the Rust AgentEvent serde-tagged enum (e.g. {"type": "text_delta", "delta": "..."}).

create_session_streaming

async with client.create_session_streaming("Hello!") as stream:
    async for event in stream:
        if event["type"] == "text_delta":
            print(event["delta"], end="", flush=True)
    result = stream.result  # WireRunResult

Accepts all the same parameters as create_session() plus preload_skills and skill_references.

start_turn_streaming

async with client.start_turn_streaming(session_id, "Follow up") as stream:
    async for event in stream:
        print(event)
    result = stream.result

Convenience methods

# Consume all events, return result
async with client.create_session_streaming("Hello!") as stream:
    result = await stream.collect()

# Accumulate text deltas
async with client.create_session_streaming("Hello!") as stream:
    full_text, result = await stream.collect_text()

Event types

Events have a type field discriminating the variant:

Type	Description
`run_started`	Agent run began
`turn_started`	New LLM turn
`text_delta`	Incremental text chunk (`delta` field)
`text_complete`	Full text available
`tool_call_requested`	Agent wants to call a tool
`tool_result_received`	Tool returned a result
`turn_completed`	LLM turn finished
`run_completed`	Agent run finished

The streaming methods are synchronous (not async def) — they return a StreamingTurn directly. The request is sent when entering the async with block.

Skills parameters

Both create_session() and start_turn() accept skill-related parameters:

# Preload skills into the system prompt at session creation
result = await client.create_session(
    "Hello!",
    preload_skills=["extraction/email", "formatting/markdown"],
)

# Inject skills for a specific turn
result = await client.start_turn(
    session_id,
    "Extract emails from this text",
    skill_references=["extraction/email"],
)

Error handling

All errors inherit from MeerkatError:

from meerkat import MeerkatError, CapabilityUnavailableError, SessionNotFoundError, SkillNotFoundError

Class	Description
`MeerkatError`	Base error with `code`, `message`, `details`, `capability_hint`
`CapabilityUnavailableError`	Required capability not available
`SessionNotFoundError`	Session not found
`SkillNotFoundError`	Skill reference cannot be resolved

Client-side error codes

Code	Description
`BINARY_NOT_FOUND`	`rkat` binary not found on PATH
`NOT_CONNECTED`	Operation attempted before `connect()`
`CONNECTION_CLOSED`	`rkat rpc` process closed unexpectedly
`VERSION_MISMATCH`	Server contract version incompatible
`CAPABILITY_UNAVAILABLE`	Capability check failed

Server-side error codes

Code	JSON-RPC code	Description
`SESSION_NOT_FOUND`	-32001	Session does not exist
`SESSION_BUSY`	-32002	Turn already in progress
`PROVIDER_ERROR`	-32010	LLM provider error
`BUDGET_EXHAUSTED`	-32011	Budget exceeded
`HOOK_DENIED`	-32012	Hook blocked operation
`AGENT_ERROR`	-32013	Internal agent error
`INVALID_PARAMS`	-32602	Invalid method parameters

Example

try:
    result = await client.start_turn("nonexistent-session-id", "hello")
except MeerkatError as e:
    print(f"Error code: {e.code}")
    print(f"Message: {e.message}")

Version compatibility

The SDK negotiates version compatibility during connect():

0.x: Requires exact minor version match (SDK 0.1.0 works with server 0.1.x)
1.0+: Requires same major version

from meerkat import CONTRACT_VERSION
print(CONTRACT_VERSION)  # "0.1.0"

Running tests

pip install -e "sdks/python[dev]"

# Type conformance tests (no rkat binary needed)
pytest sdks/python/tests/test_types.py -v

# E2E tests (requires rkat on PATH)
pytest sdks/python/tests/test_e2e.py -v

# E2E smoke tests (requires rkat + ANTHROPIC_API_KEY)
pytest sdks/python/tests/test_e2e_smoke.py -v

Rust

Python

TypeScript

Python SDK reference

Wire types

CapabilityChecker

SkillHelper

Streaming API

create_session_streaming

start_turn_streaming

Convenience methods

Event types

Skills parameters

Error handling

Client-side error codes

Server-side error codes

Example

Version compatibility

Running tests

See also

Rust

Python

TypeScript

​Wire types

​CapabilityChecker

​SkillHelper

​Streaming API

​create_session_streaming

​start_turn_streaming

​Convenience methods

​Event types

​Skills parameters

​Error handling

​Client-side error codes

​Server-side error codes

​Example

​Version compatibility

​Running tests

​See also

Wire types

CapabilityChecker

SkillHelper

Streaming API

create_session_streaming

start_turn_streaming

Convenience methods

Event types

Skills parameters

Error handling

Client-side error codes

Server-side error codes

Example

Version compatibility

Running tests

See also