SDK Reference

Python SDK

Full reference for the rankigi package.

Installation

bash

pip install rankigi

Requires Python 3.9+. No binary dependencies.

Configuration

Run npx @rankigi/cli init to issue a credential. The CLI writes a single RANKIGI_CREDENTIAL token plus RANKIGI_CHAIN_ID to your .env.

bash

RANKIGI_CREDENTIAL=rnk_live_cred_v1...
RANKIGI_CHAIN_ID=chn_your_chain_id

python

from dotenv import load_dotenv
load_dotenv()

from rankigi import Rankigi

# Reads RANKIGI_CREDENTIAL (preferred single-token form)
# and RANKIGI_CHAIN_ID from your environment.
rk = Rankigi.from_env()

# Or from a single credential token explicitly:
# rk = Rankigi.from_credential("rnk_live_cred_v1.your_token")

# Advanced / legacy: explicit four-variable form.
# rk = Rankigi(
#     api_key=os.environ["RANKIGI_API_KEY"],
#     agent_id=os.environ["RANKIGI_AGENT_ID"],
#     passport_id=os.environ["RANKIGI_PASSPORT_ID"],
#     signing_private_key=os.environ["RANKIGI_SIGNING_KEY"],
# )

Scoping events to a chain

Every event the SDK emits is scoped to a chain. Rankigi.from_env() reads RANKIGI_CHAIN_ID automatically. You can also pass chain_id to the constructor.

python

# .env
# RANKIGI_CREDENTIAL=rnk_live_cred_v1...
# RANKIGI_CHAIN_ID=chn_your_chain_id

from rankigi import Rankigi

# from_env() picks up RANKIGI_CHAIN_ID automatically.
rk = Rankigi.from_env()

# Or pass chain_id explicitly:
# rk = Rankigi(
#     api_key=...,
#     agent_id=...,
#     passport_id=...,
#     signing_private_key=...,
#     chain_id="chn_your_chain_id",
# )

Methods

`wrap_anthropic(label, args, fn)`

Wrap an Anthropic Messages call. The label, args, and result are hashed and chained.

python

response = rk.wrap_anthropic(
    "contract-review",
    {"prompt": prompt, "model": "claude-sonnet-4-20250514"},
    lambda: client.messages.create(model=model, max_tokens=1024, messages=messages),
)

`wrap_openai(label, args, fn)`

Wrap an OpenAI Chat Completions call.

python

completion = rk.wrap_openai(
    "answer-generation",
    {"model": "gpt-4o", "messages": messages},
    lambda: openai.chat.completions.create(model="gpt-4o", messages=messages),
)

`wrap(label, args, fn)`

Wrap any callable with the same governed-call pattern.

python

result = rk.wrap(
    "custom-step",
    {"input": input_data},
    lambda: do_work(input_data),
)

`track_tool_call(tool, input, output)`

Record a tool invocation. Input and output are SHA-256 hashed before transmission.

python

rk.track_tool_call(
    "stripe_refund",
    {"charge_id": "ch_abc", "amount": 100},
    {"success": True, "refund_id": "re_abc123"},
)

`track_agent_output(output)`

Record an agent output as a chain event.

python

rk.track_agent_output({
    "report_id": "rpt_456",
    "pages": 12,
})

`track_error(error)`

Record an error as a chain event.

python

try:
    run_step()
except Exception as err:
    rk.track_error(err)
    raise

`with_session(id) / with_run(id)`

Scope a block of work to a session or run identifier.

python

with rk.with_session("sess_abc123"):
    with rk.with_run("run_xyz789"):
        rk.track_tool_call("web_search", input_data, output_data)

`close()`

Flush pending events and shut down the daemon thread.

python

# Flush pending events and shut down the daemon thread.
rk.close()

Failure behavior

The SDK is non-blocking. Ingest is queued to a background daemon thread. On failure the event is logged to stderr and dropped, your agent never sees an exception from RANKIGI. The Python SDK does not retry. For production paths that require retry semantics, use the Node SDK. Enable debug=True to see verbose transport logs.

TransportDaemon thread queue (non-blocking)

Retry strategyNone (use Node SDK for retries)

Failure behaviorLogged to stderr, dropped (never raised)

Debug logsdebug=True

Full Example

python

import os
from dotenv import load_dotenv
load_dotenv()

from rankigi import Rankigi
from anthropic import Anthropic

rk = Rankigi.from_env()
anthropic = Anthropic()

def review_contract(text: str):
    return rk.wrap_anthropic(
        "contract-review",
        {"text": text, "model": "claude-sonnet-4-20250514"},
        lambda: anthropic.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{"role": "user", "content": text}],
        ),
    )

def refund_charge(charge_id: str, amount: int):
    try:
        result = stripe.refunds.create(charge=charge_id, amount=amount)
        rk.track_tool_call(
            "stripe_refund",
            {"charge_id": charge_id, "amount": amount},
            {"success": True, "refund_id": result.id},
        )
        return result
    except Exception as err:
        rk.track_error(err)
        raise

if __name__ == "__main__":
    try:
        review_contract("...")
    finally:
        rk.close()

Error Handling

python

from dotenv import load_dotenv
load_dotenv()

from rankigi import Rankigi

rk = Rankigi.from_env()

# The SDK is non-blocking. Ingest is queued to a background
# daemon thread. On failure the event is logged to stderr
# and dropped. No retries. Use the Node SDK in production
# paths that require retry semantics.
#
# For visibility, construct with debug=True:
#
# rk = Rankigi(
#     api_key=...,
#     agent_id=...,
#     passport_id=...,
#     signing_private_key=...,
#     debug=True,
# )