SDK Reference
Packages
| Package | Description | Install |
|---|---|---|
substr8 | CLI + core platform | pip install substr8 |
substr8-core | RunProof schemas + verification | pip install substr8-core |
substr8-langgraph | LangGraph instrumentation | pip install substr8-langgraph |
substr8-langgraph
The primary SDK for wrapping LangGraph agents.
Installation
pip install substr8-langgraphinstrument_graph()
Wrap a compiled LangGraph with RunProof instrumentation.
from langgraph.graph import StateGraph
from substr8_langgraph import instrument_graph
# Build and compile your graph
graph = StateGraph(MyState)
# ... add nodes and edges ...
compiled = graph.compile()
# Wrap with instrumentation
instrumented = instrument_graph(
compiled,
agent_id="my-agent",
project="my-project",
)
# Invoke
result = instrumented.invoke({"input": "..."})Parameters:
| Parameter | Type | Description |
|---|---|---|
graph | CompiledGraph | The compiled LangGraph |
agent_id | str | Unique identifier for the agent |
project | str | Project name for organization |
parent_run_id | str (optional) | Parent run ID for nested workflows |
auto_publish | bool | Auto-publish proofs to registry |
publish_url | str (optional) | Registry URL |
Returns:
The result dict includes additional proof metadata:
| Key | Description |
|---|---|
run_id | Unique run identifier |
proof | The complete RunProof object |
proof_id | Hash of the proof |
proof_status | "verified" if valid |
Example
from langgraph.graph import StateGraph, END
from typing import TypedDict
from substr8_langgraph import instrument_graph
import json
class MyState(TypedDict):
query: str
result: str
def process(state):
return {"result": f"Processed: {state['query']}"}
# Build graph
graph = StateGraph(MyState)
graph.add_node("process", process)
graph.set_entry_point("process")
graph.add_edge("process", END)
compiled = graph.compile()
# Instrument
instrumented = instrument_graph(compiled, agent_id="example", project="demo")
# Run
result = instrumented.invoke({"query": "Hello"})
# Save proof
with open("runproof.json", "w") as f:
json.dump(result["proof"].model_dump(mode="json"), f, indent=2)
print(f"Run ID: {result['run_id']}")
print(f"Status: {result['proof_status']}")substr8-core
Core schemas and verification logic.
Installation
pip install substr8-coreRunProof
The RunProof Pydantic model:
from substr8_core import RunProof
# Load from file
import json
with open("runproof.json") as f:
data = json.load(f)
proof = RunProof.model_validate(data)
print(proof.header.proof_id)
print(proof.header.agent_id)
print(len(proof.trace))verify_runproof()
Verify a RunProof programmatically:
from substr8_core import verify_runproof
result = verify_runproof(proof_dict)
if result.valid:
print("✓ Proof is valid")
else:
print(f"✗ Invalid: {result.errors}")Returns:
class VerificationResult:
valid: bool
errors: list[str]
checks: dict[str, bool] # Individual check resultsRunState
For building proofs manually:
from substr8_core import RunState, RunStatus
state = RunState(agent_id="my-agent", project="my-project")
# Start
state.start()
# Add events
state.add_event("tool_called", {"tool": "search", "query": "..."})
state.add_event("tool_result", {"result": "..."})
# End
state.end(RunStatus.COMPLETED, outputs={"answer": "..."})
# Build proof
proof = state.build_proof()substr8 CLI
The CLI is also importable:
from substr8.runproof.v2 import verify as verify_cli
import json
with open("runproof.json") as f:
proof = json.load(f)
result = verify_cli(proof)
print(f"Valid: {result.valid}")Type Definitions
TraceEvent
class TraceEvent(BaseModel):
type: str
timestamp: datetime
entry_hash: str
prev_hash: str | None
# ... event-specific fieldsCommitments
class Commitments(BaseModel):
event_root: str # Merkle root
proof_hash: str # Hash of proof
signature: SignatureSignature
class Signature(BaseModel):
algorithm: str # "ed25519"
public_key: str # Public key
value: str # Signature bytesNext Steps
- Quickstart — Full example
- RunProof — Schema details
- CLI Reference — Command line tools