Skip to main content

Python SDK

Official Python SDK for Shim. One dependency (httpx). Sync and async clients included.

Installation

pip
pip install shim-sdk
poetry
poetry add shim-sdk
uv
uv add shim-sdk

Quick Start

from shim import ShimClient

with ShimClient(api_key="sk_live_...") as shim:
    result = shim.repair(raw_output='{"name": "John", "age": 30')

    if result.success:
        print(result.repaired)  # {'name': 'John', 'age': 30}

Client Configuration

Sync

from shim import ShimClient

shim = ShimClient(
    api_key="sk_live_...",   # Required
    base_url="https://api.shim.so"  # Optional, default
)

# Always close when done, or use as context manager
shim.close()

Async

from shim import AsyncShimClient

shim = AsyncShimClient(
    api_key="sk_live_...",
    base_url="https://api.shim.so"
)

await shim.close()
Parameters:
  • api_key (str, required): Your Shim API key
  • base_url (str, optional): API base URL. Default: https://api.shim.so

Batch Repair

Sync

from shim import ShimClient

with ShimClient(api_key="sk_live_...") as shim:
    result = shim.repair(raw_output='{"name": "John", "age": "30"')

    print(result.success)              # True
    print(result.repaired)             # {'name': 'John', 'age': '30'}
    print(result.metadata.confidence)  # 'high'

Async

from shim import AsyncShimClient

async with AsyncShimClient(api_key="sk_live_...") as shim:
    result = await shim.repair(raw_output='{"name": "John", "age": "30"')
    print(result.repaired)

With Schema Validation

from shim import ShimClient

schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age":  {"type": "number"}
    },
    "required": ["name", "age"]
}

with ShimClient(api_key="sk_live_...") as shim:
    result = shim.repair(
        raw_output='{"name": "John", "age": "30"',
        schema=schema,
        mode="strict"
    )

    if result.success:
        print(result.repaired)             # {'name': 'John', 'age': 30}
        print(result.metadata.confidence)  # 'medium' — schema repair present

Streaming Repair

stream.start()

Start a new streaming session. Schema and mode are set here — not on push.
session = shim.stream.start(
    schema={"type": "object", "properties": {"name": {"type": "string"}}},
    mode="strict"
)

print(session.session_id)  # 'sess_abc123'
print(session.expires_at)  # unix timestamp, 60s from now

stream.push()

Push a chunk. Returns the current streaming state.
state = shim.stream.push(session.session_id, '{"name": "Jo')

print(state.structurally_complete)  # False
print(state.buffered)               # '{"name": "Jo'
print(state.safe_to_emit)           # False

stream.finalize()

Close the session. Returns the full repair response.
result = shim.stream.finalize(session.session_id)
print(result.repaired)  # {'name': 'John'}

Complete Streaming Example

Sync

from shim import ShimClient

with ShimClient(api_key="sk_live_...") as shim:
    session = shim.stream.start()

    # Push chunks as they arrive from LLM
    shim.stream.push(session.session_id, '{"name": "Jo')
    shim.stream.push(session.session_id, 'hn", "age": 30')

    # Finalize when stream ends
    result = shim.stream.finalize(session.session_id)
    print(result.repaired)  # {'name': 'John', 'age': 30}

Async

from shim import AsyncShimClient

async with AsyncShimClient(api_key="sk_live_...") as shim:
    session = await shim.stream.start()

    await shim.stream.push(session.session_id, '{"name": "Jo')
    await shim.stream.push(session.session_id, 'hn", "age": 30')

    result = await shim.stream.finalize(session.session_id)
    print(result.repaired)  # {'name': 'John', 'age': 30}

Type Hints

The SDK ships with full type definitions. All response objects are dataclasses.
from shim import ShimClient, RepairResponse, StreamingState

with ShimClient(api_key="sk_live_...") as shim:
    result: RepairResponse = shim.repair(raw_output="{}")

    # Full autocomplete on all fields
    print(result.metadata.confidence)       # 'high' | 'medium' | 'low' | 'n/a'
    print(result.metadata.syntax_repairs)   # list[RepairDetail]
    print(result.metadata.warnings)         # list[Warning]
    print(result.metadata.errors)           # list[RepairError]
Exported types:
TypeDescription
RepairResponseTop-level batch/finalize response
RepairMetadataConfidence, repairs, warnings, errors
RepairDetailSingle repair operation (type, confidence, field)
WarningNon-critical issue
RepairErrorCritical failure with recoverability flag
StreamSessionSession ID + expiry from stream.start()
StreamingStateIncremental parse state from stream.push()

Error Handling

The SDK raises httpx.HTTPStatusError for transport-level failures. Shim itself always returns HTTP 200 — application errors are in the response body.
import httpx
from shim import ShimClient

with ShimClient(api_key="sk_live_...") as shim:
    try:
        result = shim.repair(raw_output="not json at all")

        if not result.success:
            for error in result.metadata.errors:
                print(f"{error.code}: {error.message}")
                print(f"  recoverable: {error.recoverable}")
    except httpx.HTTPStatusError as e:
        print(f"Transport error: {e.response.status_code}")

Next Steps

Batch Endpoint

API reference for batch repair

Streaming Endpoints

API reference for streaming repair

Response Object

Full response schema

TypeScript SDK

Node.js / browser SDK