Python SDK reference

Wire types, capability checking, skill helpers, event streaming, and error handling for the Python SDK.

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

---

See also

• Python SDK overview - getting started and MeerkatClient API
• TypeScript SDK - TypeScript SDK
• RPC reference - JSON-RPC protocol specification