SDK Reference

Python SDK

Full reference for the rankigi package.

Installation

bash

pip install rankigi

Requires Python 3.9+. No binary dependencies.

Configuration

python

import os
from rankigi import Rankigi

client = Rankigi(
    api_key=os.environ["RANKIGI_API_KEY"],
    agent_id=os.environ["RANKIGI_AGENT_ID"],
)

Get your API key from the Dashboard → API Keys page.

Methods

`track_tool_call()`

Record an agent tool invocation with input/output.

python

client.track_tool_call(
    tool="stripe_refund",
    input={"amount": 100, "reason": "duplicate_charge"},
    output={"success": True, "refund_id": "re_abc123"},
)

`track_agent_output()`

Record a general agent action or output.

python

client.track_agent_output(
    action="generate_report",
    output={"report_id": "rpt_456", "pages": 12},
    severity="info",
)

`track_error()`

Record an error during agent operation.

python

client.track_error(
    tool="database_query",
    error="Connection timeout after 30s",
    severity="critical",
)

`track_custom_event()`

Record any custom event with arbitrary payload.

python

client.track_custom_event(
    action="policy_decision",
    tool="approval_engine",
    payload={
        "decision": "approved",
        "amount": 5000,
        "approver": "agent_7b2f",
    },
    severity="warn",
)

`initiate_handshake() / complete_handshake() / verify_handshake()`

Establish and verify cryptographic handshakes between agents.

python

# Agent A initiates a handshake with Agent B
handshake = client.initiate_handshake(
    target_agent_id="agent-b-uuid",
    purpose="data_exchange",
    metadata={"dataset": "customer_records"},
)

# Agent B completes the handshake
client.complete_handshake(
    handshake_id=handshake["id"],
    accepted=True,
)

# Either party can verify
result = client.verify_handshake(
    handshake_id=handshake["id"],
)

`flush() / close()`

Force flush buffered events or gracefully shut down the client.

python

# Force flush all buffered events immediately
client.flush()

# Gracefully close the client
client.close()

Context Manager

The Python SDK supports the context manager protocol for automatic cleanup.

python

import os
from rankigi import Rankigi

# Context manager automatically flushes and closes on exit
with Rankigi(
    api_key=os.environ["RANKIGI_API_KEY"],
    agent_id=os.environ["RANKIGI_AGENT_ID"],
) as client:
    client.track_tool_call(
        tool="send_email",
        input={"to": "user@example.com", "subject": "Invoice"},
        output={"status": "sent", "message_id": "msg_xyz"},
    )
# client.close() called automatically

Async Support

For asyncio-based applications, use AsyncRankigi.

python

import os
from rankigi import AsyncRankigi

client = AsyncRankigi(
    api_key=os.environ["RANKIGI_API_KEY"],
    agent_id=os.environ["RANKIGI_AGENT_ID"],
)

async def process_refund(charge_id: str, amount: int):
    result = await stripe.refunds.create(
        charge=charge_id,
        amount=amount,
    )

    await client.track_tool_call(
        tool="stripe_refund",
        input={"charge_id": charge_id, "amount": amount},
        output={"success": True, "refund_id": result.id},
    )

    return result

# Async context manager also supported
async with AsyncRankigi(
    api_key=os.environ["RANKIGI_API_KEY"],
    agent_id=os.environ["RANKIGI_AGENT_ID"],
) as client:
    await client.track_tool_call(...)

Buffer and Retry Behavior

The SDK uses a daemon thread to transport events to the RANKIGI API. This ensures your agent is never blocked by governance operations.

TransportDaemon thread (non-blocking)

Flush interval5,000ms (configurable)

Max buffer size100 events (configurable)

Retry strategyExponential backoff, 3 retries

Failure behaviorSilent drop (sidecar principle)

Full Example

python

import os
from rankigi import Rankigi

def main():
    client = Rankigi(
        api_key=os.environ["RANKIGI_API_KEY"],
        agent_id=os.environ["RANKIGI_AGENT_ID"],
    )

    try:
        # Your agent logic
        result = process_customer_request(request)

        # Track the action
        client.track_tool_call(
            tool="customer_service",
            input={"request_id": request.id},
            output={
                "resolution": result.resolution,
                "actions_taken": result.actions,
            },
        )

        # Track any follow-up decisions
        if result.requires_escalation:
            client.track_custom_event(
                action="escalation",
                payload={
                    "reason": result.escalation_reason,
                    "priority": "high",
                },
                severity="warn",
            )

    except Exception as e:
        client.track_error(
            tool="customer_service",
            error=str(e),
            severity="critical",
        )
        raise

    finally:
        client.close()

if __name__ == "__main__":
    main()

Error Handling

python

from rankigi import Rankigi, RankigiError

client = Rankigi(
    api_key=os.environ["RANKIGI_API_KEY"],
    agent_id=os.environ["RANKIGI_AGENT_ID"],
)

# The SDK uses a daemon thread for event transport.
# It never raises exceptions in your agent's main thread.
# Events are buffered and retried silently.

# To handle errors explicitly, set an error callback:
def on_error(error: RankigiError):
    print(f"[rankigi] {error.code}: {error.message}")
    # error.code: "AUTH_FAILED" | "RATE_LIMITED" | "NETWORK_ERROR"
    # error.status_code: HTTP status code (if applicable)

client.on_error = on_error