sigmodx-integrations

Agent framework integrations for Sigmodx — audit infrastructure for AI agents making consequential decisions.

Installation

pip install sigmodx-integrations

LangChain (Live)

Register one callback. Every agent tool call is automatically logged to Sigmodx with cryptographic attestation.

from sigmodx import SigmodxClient
from sigmodx_integrations import SigmodxCallbackHandler, ANOMALY_DETECTION_CONFIG

client = SigmodxClient(
    api_key="your-api-key",
    agent_id="your-agent-uuid",
)

handler = SigmodxCallbackHandler(
    client=client,
    config=ANOMALY_DETECTION_CONFIG,
)

result = agent_executor.invoke(
    {"input": "Check transaction TXN-2026-4421"},
    config={"callbacks": [handler]},
)

The handler:

• Hashes tool inputs client-side (your data never leaves your environment)
• Extracts decision type, rationale, and metadata from tool output
• Submits the decision to Sigmodx's append-only audit trail
• Never blocks agent execution — errors are logged, not raised

LangGraph (Live)

Two integration patterns — use whichever fits your graph structure.

Pattern 1: Stream event handler (recommended)

Process events from .astream_events() to log all tool calls automatically.

from sigmodx import SigmodxClient
from sigmodx_integrations.langgraph import (
    SigmodxLangGraphCallback,
    LANGGRAPH_ANOMALY_CONFIG
)

client = SigmodxClient(api_key="...", agent_id="...")
handler = SigmodxLangGraphCallback(
    client=client,
    config=LANGGRAPH_ANOMALY_CONFIG
)

# Process stream events
async for event in graph.astream_events(inputs, version="v2"):
    await handler.aprocess_event(event)

# Or process synchronously
for event in graph.stream(inputs):
    handler.process_event(event)

Pattern 2: Node decorator

Wrap specific decision-making nodes directly.

from sigmodx_integrations.langgraph import (
    sigmodx_node,
    LANGGRAPH_ANOMALY_CONFIG
)

@sigmodx_node(client=client, config=LANGGRAPH_ANOMALY_CONFIG)
def analyze_transaction(state: dict) -> dict:
    # your existing node logic
    return {
        "decision": "flag",
        "rationale": "Amount 3x historical average.",
        "anomaly_subtype": "unusual_amount",
        "severity": "high",
        "transaction_amount": state["amount"]
    }

graph.add_node("analyze_transaction", analyze_transaction)

Installation

pip install "sigmodx-integrations[langgraph]"
# or
pip install sigmodx-integrations langgraph

CrewAI (Live)

Two patterns: task-level callback or crew-level step callback.

Task callback (recommended for specific tasks)

from crewai import Task
from sigmodx import SigmodxClient
from sigmodx_integrations.crewai import (
    SigmodxTaskCallback,
    CREWAI_ANOMALY_CONFIG
)

client = SigmodxClient(api_key="...", agent_id="...")
sigmodx_callback = SigmodxTaskCallback(
    client=client,
    config=CREWAI_ANOMALY_CONFIG,
    task_inputs={"txn_ref": "TXN-001", "amount": 5000}
)

task = Task(
    description="Analyze transaction for anomalies",
    agent=analyst_agent,
    expected_output="flag/clear/escalate with rationale",
    callback=sigmodx_callback
)

Step callback (all agent steps)

from crewai import Crew
from sigmodx_integrations.crewai import SigmodxStepCallback

crew = Crew(
    agents=[analyst, compliance],
    tasks=[analysis_task, decision_task],
    step_callback=SigmodxStepCallback(client=client,
                                       config=CREWAI_ANOMALY_CONFIG)
)

Install

pip install "sigmodx-integrations[crewai]"

OpenAI Agents SDK (Live)

Implement RunHooks and pass to Runner.run().

from agents import Agent, Runner
from sigmodx import SigmodxClient
from sigmodx_integrations.openai_agents import (
    SigmodxRunHooks,
    OPENAI_ANOMALY_CONFIG
)

client = SigmodxClient(api_key="...", agent_id="...")
hooks = SigmodxRunHooks(client=client, config=OPENAI_ANOMALY_CONFIG)

agent = Agent(
    name="AnomalyDetector",
    instructions="Detect financial anomalies.",
    tools=[check_transaction, flag_anomaly]
)

result = await Runner.run(agent, "Check TXN-2026-4421", hooks=hooks)

Install

pip install "sigmodx-integrations[openai-agents]"

Custom scenario mapping

from sigmodx_integrations import SigmodxCallbackHandler, ScenarioConfig

config = ScenarioConfig(
    scenario="invoice_approval",
    filter_tools=["approve_invoice", "reject_invoice"],
    decision_type_extractor=lambda output: output.get("decision"),
    rationale_extractor=lambda output: output.get("reason"),
    metadata_extractor=lambda output: {
        "invoice_amount": output.get("amount"),
        "vendor_id": output.get("vendor_id"),
    },
)

handler = SigmodxCallbackHandler(client=client, config=config)

Universal adapter (any framework)

from sigmodx_integrations import SigmodxAdapter

adapter = SigmodxAdapter(client=client, scenario="anomaly_detection")

adapter.log(
    inputs={"txn_ref": "TXN-001", "amount": 5000},
    decision_type="flag",
    rationale="Amount 3x historical average.",
    anomaly_subtype="unusual_amount",
    severity="high",
    transaction_amount=5000,
)

Supported frameworks

Request framework prioritization: github.com/Sigmodx/integrations-python/issues

Links

• Sigmodx
• Agent API reference
• GitHub