BlocklogAI Compliance
Documentation Topics
OverviewQuickstartCore ConceptsPython SDKTypeScript SDKREST API ReferenceIncident ReconstructionVPC DeploymentOpenTelemetry Integration
SDK Reference

Python SDK Reference

Method signatures, parameter schemas, return types, runnable code examples, and framework integrations for the Blocklog Python client.

Method

blocklog.init()

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

signature
python
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

Parameters

  • api_key · str, optional

    Key for verification. Falls back to BLOCKLOG_API_KEY.

  • base_url · str, optional

    Custom API endpoint override. Falls back to BLOCKLOG_BASE_URL.

  • signing_key · str, optional

    HMAC-SHA256 key for tamper-evident log signatures. Falls back to BLOCKLOG_SDK_SIGNING_KEY.

  • timeout · float, optional

    Request timeout in seconds. Default is 10.

  • max_retries · int, optional

    Ingestion retries. Default is 3.

Example

example.py
python
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.

Note — Class decoration only emits AGENT_START. For full lifecycle tracing, decorate the specific method (e.g. run, execute) rather than the class itself.
signature
python
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]

Parameters

  • name · str, optional

    Human-readable agent name. Defaults to func.__name__.

  • version · str

    Semver-style version string. Default is "1.0".

  • tags · list[str], optional

    Optional list of string tags.

  • metadata · dict, optional

    Arbitrary extra data attached to the agent session.

Example

example.py
python
@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.

signature
python
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]

Parameters

  • name · str, optional

    Human-readable tool name. Defaults to func.__name__.

  • schema · dict, optional

    Optional dict describing the input schema.

  • tags · list[str], optional

    Optional string tags.

  • metadata · dict, optional

    Arbitrary extra data.

Example

example.py
python
@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 an AI decision.

signature
python
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]

Parameters

  • type · str

    Decision type identifier (e.g. "CREDIT_APPROVAL").

  • asset · str, optional

    Asset or entity this decision evaluates.

  • confidence · float, optional

    Model confidence score from 0.0 to 1.0.

  • metadata · dict, optional

    Extra structured audit fields.

  • agent_id · str, optional

    Links this decision to a specific agent session.

  • trace_id · str, optional

    Links this decision to a specific trace.

Returns: DecisionContext

  • record_input(**kwargs)

    Record structured inputs. Returns self.

  • record_output(**kwargs)

    Record structured outputs. Returns self.

  • tag(*tags)

    Attach string labels. Returns self.

  • request_approval(reason, reviewer=None)

    Non-blocking request for human approval. Returns self.

  • verify()

    Verify the decision after the with block exits.

Example

example.py
python
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 and Slack notifications asynchronously.

signature
python
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]

Parameters

  • decision_id · str, optional

    ID of the decision this approval request relates to.

  • reason · str

    Why human review is required.

  • reviewer · str, optional

    Email or identifier of the assigned reviewer.

  • log_id · str, optional

    ID of a specific log entry if not tied to a decision.

  • metadata · dict, optional

    Arbitrary extra audit data.

Example

example.py
python
blocklog.approval.request(
    decision_id="dec_123",
    reason="Price exceeds risk 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.

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

Parameters

  • trace_id · str

    Trace ID to reconstruct.

  • token_id · str, optional

    Restrict the replay to a specific token within the trace.

  • metadata · dict, optional

    Extra context attached to the replay session.

Example

example.py
python
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.

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

Parameters

  • log_id · str

    ID of the log entry to verify.

Example

example.py
python
result = blocklog.verify.log("log-uuid")
assert result["status"] == "verified"
Framework Instrumentation

Framework Integrations

Built-in instrumentation for LangChain, LangGraph, the OpenAI SDK, and LiteLLM. Each integration wires into native hooks and emits structured audit events to the Blocklog ingest API.

Common Behavior Across All Integrations

  • —Concurrent run safety — LangChain/LangGraph key state by run_id, LiteLLM by litellm_call_id, and OpenAI by a ContextVar, so parallel executions never clobber each other.
  • —Context inheritance — trace_id, session_id, and agent_id are pulled automatically from active Blocklog context.
  • —Safe serialization — Pydantic v1 (.dict()) and v2 (.model_dump()) models, plain Python types, and arbitrary objects are safely handled.
  • —Error events always fire — on any exception, an errored event is emitted before propagation so audit trails are never incomplete.
Integration

LangChain

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

Setup

setup.py
python
import blocklog

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

Usage

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

Events Emitted

EventCausality typeFired when
agent.chain.startedchain_startA chain begins; includes inputs and input_keys.
agent.chain.completedchain_endA chain finishes successfully; includes outputs.
agent.chain.erroredchain_errorA chain raises an exception.
agent.model.startedllm_startAn LLM call begins; includes serialized model info and prompts.
agent.model.completedllm_endAn LLM call finishes; includes the full response.
agent.model.erroredllm_errorAn LLM call raises an exception.
agent.tool.startedtool_startA tool begins execution; includes the raw input string.
agent.tool.completedtool_endA tool finishes; includes output.
agent.tool.erroredtool_errorA tool raises an exception.
  • —Every event carries a span_id, a parent_run_id for causality linking, and an agent_metadata block with framework metadata.
Integration

LangGraph

Extends the LangChain handler pattern with graph-native event families. Inspects the serialized payload to emit graph, node, and subgraph events accurately.

Setup

setup.py
python
import blocklog

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

Usage

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

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

Events Emitted

Graph lifecycle

EventCausality typeFired when
agent.graph.startedgraph_startThe top-level CompiledGraph begins; includes graph_name, inputs, and input_keys.
agent.graph.completedgraph_endThe graph finishes successfully; includes outputs.
agent.graph.erroredgraph_errorThe graph raises an unhandled exception.

Node lifecycle

EventCausality typeFired when
agent.graph.node.startednode_startA node begins; includes node_name and inputs.
agent.graph.node.completednode_endA node finishes; includes node_name and outputs.
agent.graph.node.errorednode_errorA node raises an exception; includes node_name and error details.

Subgraph lifecycle

EventCausality typeFired when
agent.graph.subgraph.startedsubgraph_startA nested subgraph begins; includes subgraph_name and inputs.
agent.graph.subgraph.completedsubgraph_endA nested subgraph finishes; includes outputs.

Checkpoint lifecycle

EventCausality typeFired when
agent.graph.checkpoint.startedcheckpoint_startLangGraph is about to write a checkpoint; includes thread_id and checkpoint_ns.
agent.graph.checkpoint.completedcheckpoint_endA checkpoint is persisted; includes checkpoint_id, thread_id, and checkpoint_ns.
  • —All agent.model.* and agent.tool.* events are also emitted for calls made inside nodes.
  • —State dicts are safely serialized before emission.
Integration

OpenAI SDK

Instruments create methods on OpenAI resource classes directly. Every OpenAI(), AsyncOpenAI(), and AzureOpenAI() instance is instrumented automatically.

Setup

setup.py
python
import blocklog

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

Instrument a single instance instead

usage.py
python
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)

Events Emitted

EventCausality typeFired when
agent.model.startedllm_startA chat.completions.create or responses.create call begins; includes model, messages, and request parameters.
agent.model.completedllm_endThe call finishes; includes the full response, token usage, duration_s, and streamed flag.
agent.model.erroredllm_errorThe call raises an exception.
  • —Streaming is handled transparently — chunks are re-yielded unmodified, and agent.model.completed fires once the stream is fully consumed.
  • —Async create methods are wrapped with an async-native path automatically.
Integration

LiteLLM

Registers via litellm.callbacks using LiteLLM's CustomLogger base class with no monkey-patching required.

Setup

setup.py
python
import blocklog

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

Usage

usage.py
python
import litellm

# Replace or append to existing callbacks
litellm.callbacks = [handler]

Events Emitted

EventCausality typeFired when
agent.model.pre_callllm_pre_callBefore the HTTP request is dispatched; includes model, messages, stream, and call ID.
agent.model.post_callllm_post_callAfter the API responds, before the success/failure split.
agent.model.stream_chunkllm_stream_chunkFor each individual chunk on the sync streaming path.
agent.model.completedllm_endA call finishes successfully (sync or async); includes response, usage, and cost.
agent.model.erroredllm_errorA call fails (sync or async); includes error_type and duration_s.
  • —span_id is derived from litellm_call_id for consistent tracing across pre-call, post-call, and success/failure hooks.