Python SDK Reference

Method

blocklog.init()

Initialize the global SDK configuration. Call once at application startup.

def init(
    api_key: str | None = None,
    *,
    base_url: str | None = None,
    signing_key: str | None = None,
    timeout: float | None = None,
    max_retries: int | None = None,
    debug: bool = False,
) -> BlocklogClient

import blocklog
blocklog.init(api_key="blk_live_xxxx", debug=True)

Decorator

@blocklog.agent

Trace an AI agent function, linking it to a Blocklog session. Handles sync and async functions automatically.

def agent(
    func: F | None = None,
    *,
    name: str | None = None,
    version: str = "1.0",
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
) -> F | Callable[[F], F]

@blocklog.agent(name="stock-trader", version="2.0", tags=["prod"])
def run_trading(ticker):
    pass

Decorator

@blocklog.tool

Wrap a helper function or external API call to record it as a TOOL_CALL event. Inherits trace context from the surrounding @agent.

def tool(
    func: F | None = None,
    *,
    name: str | None = None,
    schema: dict[str, Any] | None = None,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
) -> F | Callable[[F], F]

@blocklog.tool(name="fetch-price", tags=["market-data"])
def get_price(ticker):
    return 189.30

Context Manager

blocklog.decision()

Define a block for recording structured inputs, outputs, and review tasks associated with a system choice.

def decision(
    *,
    type: str,
    asset: str | None = None,
    confidence: float | None = None,
    metadata: dict[str, Any] | None = None,
    agent_id: str | None = None,
    trace_id: str | None = None,
) -> Generator[DecisionContext, None, None]

with blocklog.decision(type="BUY", asset="AAPL", confidence=0.9) as d:
    d.record_input(price=189.30)
    d.tag("high-conviction")
    d.record_output(order_id="ord_99")

Governance Method

blocklog.approval.request()

Request human review for an action. Triggers webhooks asynchronously.

def request(
    decision_id: str | None = None,
    *,
    reason: str,
    reviewer: str | None = None,
    log_id: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]

blocklog.approval.request(
    decision_id="dec_123",
    reason="Price exceeds limits",
    reviewer="cro@company.com"
)

• —Related: blocklog.approval.reject(), escalate(), list_overrides(), and audit_trail() round out the human-in-the-loop workflow.

Forensic Method

blocklog.replay()

Instantiate a session to examine execution traces, staleness heatmaps, and counterfactual outcomes.

def replay(
    trace_id: str,
    *,
    token_id: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> ReplaySession

session = blocklog.replay("trace-uuid")
cause = session.root_cause()
print(cause["description"])

Verification Method

blocklog.verify.log()

Confirm a single log entry has not been tampered with since creation.

def log(log_id: str) -> dict[str, Any]

result = blocklog.verify.log("log-uuid")
assert result["status"] == "verified"

Integration

LangChain

Attach a callback handler to any chain, agent, or LLM to capture chain, model, and tool lifecycle events.

import blocklog

client = blocklog.init(api_key="blk_...")
handler = client.instrument_langchain()

chain.invoke(inputs, config={"callbacks": [handler]})

• —Every event carries a span_id (derived from run_id), a parent_run_id for causality linking, and an agent_metadata block with framework, captured_at, and any LangChain-provided metadata.

Integration

LangGraph

Extends the LangChain handler pattern with graph-native event families. Because LangGraph routes both top-level graphs and individual nodes through on_chain_*, the handler inspects the serialized payload to tell them apart and emit the right event.

import blocklog

client = blocklog.init(api_key="blk_...")
handler = client.instrument_langgraph()

result = graph.invoke(state, config={"callbacks": [handler]})

# Or via RunnableConfig
from langchain_core.runnables import RunnableConfig
result = graph.invoke(state, config=RunnableConfig(callbacks=[handler]))

• —All agent.model.* and agent.tool.* events are also emitted for calls made inside nodes, identical to the LangChain integration.
• —A _node_name_stack keyed by run_id ensures node names are correctly attributed on completion and error callbacks, which LangGraph does not re-supply.
• —State dicts — which often contain Pydantic models — are safely serialized before emission.

Integration

OpenAI SDK

The OpenAI Python SDK has no callback system, so instrumentation works by monkey-patching the create method on the SDK's resource classes directly. Every OpenAI(), AsyncOpenAI(), and AzureOpenAI() instance in the process — including ones already constructed — is instrumented automatically.

import blocklog

client = blocklog.init(api_key="blk_...")
client.instrument_openai_agents()

import openai
from blocklog.integrations.openai_agents import instrument_openai

blocklog_client = blocklog.init(api_key="blk_...")
openai_client = openai.OpenAI(api_key="...")

instrument_openai(blocklog_client, openai_client=openai_client)

• —Streaming is handled transparently for sync and async clients — chunks are re-yielded unmodified, and agent.model.completed fires only once the stream is fully consumed, with deltas merged into one coherent response.
• —Async create methods are wrapped with an async-native path, including async streaming via async for; sync vs. async is detected automatically.
• —A contextvars.ContextVar (blocklog_openai_span_id) propagates span_id across nested or concurrent calls within the same async context.
• —A _blocklog_wrapped sentinel flag makes calling instrument_openai_agents() more than once safe.

Integration

LiteLLM

LiteLLM has a first-class CustomLogger base class. The handler subclasses it and registers via litellm.callbacks — the same pattern every other LiteLLM observability tool uses, with no monkey-patching required.

import blocklog

client = blocklog.init(api_key="blk_...")
handler = client.instrument_litellm()

import litellm

# Replace all existing callbacks
litellm.callbacks = [handler]

# Or append alongside existing callbacks
litellm.callbacks += [handler]

• —span_id is derived from litellm_call_id, stamped onto kwargs by LiteLLM before any hook fires, so pre-call, post-call, and success/failure events for one request share a consistent span ID.
• —usage (prompt_tokens, completion_tokens, total_tokens) and response_cost are promoted to the top level of agent.model.completed payloads.
• —log_success_event / log_failure_event handle sync calls; async_log_success_event / async_log_failure_event handle acompletion and async streaming — both delegate to shared handlers so payloads are identical regardless of mode.
• —Any metadata dict passed via litellm_params is forwarded into the agent_metadata block on every event.