Python SDK reference

from meerkat import RunResult

from meerkat import Session

from meerkat import DeferredSession

from meerkat import Usage

from meerkat import SessionInfo

from meerkat import Capability

# Property: list[Capability] populated during connect()
for cap in client.capabilities:
    if cap.available:
        print(f"{cap.id}: {cap.description}")

# Boolean check
if client.has_capability("shell"):
    ...

# Guarded code path — raises CapabilityUnavailableError if not available
client.require_capability("comms")

from meerkat import SkillKey

key = SkillKey(source_uuid="abc123", skill_name="code-review")

SkillRef = Union[SkillKey, str]

# Text-only (unchanged)
session = await client.create_session("Describe this project")

# Multimodal with image
session = await client.create_session([
    {"type": "text", "text": "What is in this screenshot?"},
    {"type": "image", "media_type": "image/png", "data": base64_png},
])

# Multimodal turn
result = await session.turn([
    {"type": "text", "text": "Now analyze this second image:"},
    {"type": "image", "media_type": "image/jpeg", "data": base64_jpg},
])

from meerkat import SchemaWarning

from meerkat import SkillRuntimeDiagnostics, SourceHealthSnapshot, SkillQuarantineDiagnostic

from meerkat import EventStream

# Pattern matching on typed events
async with session.stream("Explain monads") as events:
    async for event in events:
        match event:
            case TextDelta(delta=chunk):
                print(chunk, end="", flush=True)
            case ToolCallRequested(name=name, args=args):
                print(f"\nCalling tool: {name} with args: {args}")
            case TurnCompleted(usage=u):
                print(f"\nTokens: {u.input_tokens} in / {u.output_tokens} out")
            case _:
                pass
    result = events.result

# Consume silently, get RunResult
result = await session.stream("prompt").collect()

# Accumulate text, get (full_text, RunResult)
text, result = await session.stream("prompt").collect_text()

from meerkat import (
    Event,
    RunStarted, RunCompleted, RunFailed,
    TurnStarted, TurnCompleted,
    TextDelta, TextComplete,
    ToolCallRequested, ToolResultReceived,
    ToolExecutionStarted, ToolExecutionCompleted, ToolExecutionTimedOut,
    CompactionStarted, CompactionCompleted, CompactionFailed,
    BudgetWarning,
    Retrying,
    HookStarted, HookCompleted, HookFailed, HookDenied,
    HookRewriteApplied, HookPatchPublished,
    SkillsResolved, SkillResolutionFailed,
    InteractionComplete, InteractionFailed,
    StreamTruncated,
    ToolConfigChanged, ToolConfigChangedPayload,
    UnknownEvent,
)

from meerkat import MeerkatError, CapabilityUnavailableError, SessionNotFoundError, SkillNotFoundError

from meerkat import MeerkatClient, MeerkatError, CapabilityUnavailableError

async def main():
    async with MeerkatClient() as client:
        session = await client.create_session("Hello!")

        try:
            result = await session.turn("Follow up")
        except MeerkatError as e:
            print(f"Error code: {e.code}")
            print(f"Message: {e.message}")

        try:
            client.require_capability("comms")
        except CapabilityUnavailableError as e:
            print(f"Comms not available: {e.message}")

from meerkat import CONTRACT_VERSION
print(CONTRACT_VERSION)  # e.g. "0.5.0"

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