sagashield 0.1.2

ACID transactional Saga runtime, Step-0 security guardrail, and MCP server for autonomous AI agents.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

"""Realistic LangGraph/CrewAI-style pipeline protected by SagaShield.

Run from the repo root (after `maturin develop --features python`):

    python examples/python_agent_demo.py

A planner node writes an order file, a billing node charges the customer,
then a shipping node crashes (simulated carrier outage). SagaShield rolls
back charge + file automatically; the audit trail is exported as OTel JSON.
"""

import os
import sys
import tempfile

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from sagashield import SagaKernel, SecurityPolicy, transactional_tool

WORKSPACE = tempfile.mkdtemp(prefix="sagashield_demo_")
LEDGER = {}


def remove_file(ctx, args, output):
    if os.path.exists(args["path"]):
        os.remove(args["path"])
        print(f"  [compensate] deleted {args['path']}")


@transactional_tool("write_order", compensate_with=remove_file)
def write_order(ctx, args):
    with open(args["path"], "w", encoding="utf-8") as fh:
        fh.write(args["content"])
    print(f"  [node:plan] wrote {args['path']}")
    return {"path": args["path"]}


def refund_charge(ctx, args, output):
    LEDGER[args["payment_id"]] = "REFUNDED"
    print(f"  [compensate] refunded {args['payment_id']}")


@transactional_tool("charge_customer", compensate_with=refund_charge)
def charge_customer(ctx, args):
    LEDGER[args["payment_id"]] = "CHARGED"
    print(f"  [node:bill] charged {args['payment_id']} ({args['amount']}c)")
    return {"payment_id": args["payment_id"]}


@transactional_tool("ship_order")
def ship_order(ctx, args):
    # Simulated carrier outage mid-graph (LangGraph node failure).
    raise RuntimeError("carrier API 503: shipping unavailable")


def main():
    print("=== SagaShield × Python agent demo ===")
    kernel = SagaKernel(policy=SecurityPolicy([WORKSPACE]))
    kernel.register_decorated()
    session = kernel.new_session()
    print(f"saga session: {session}")

    order = os.path.join(WORKSPACE, "order-7.txt")
    kernel.begin_planning()
    kernel.begin_tool("write_order")
    kernel.execute_tool("write_order", {"path": order, "content": "7 × widgets"})
    kernel.begin_tool("charge_customer")
    kernel.execute_tool(
        "charge_customer",
        {"payment_id": "demo-pay-7", "amount": 1999},
        idempotency_key="demo-7",
    )

    kernel.begin_tool("ship_order")
    try:
        kernel.execute_tool("ship_order", {"order": order})
    except RuntimeError as exc:
        print(f"[graph] node failed, kernel rolled back: {exc}")

    print(f"[graph] ledger={LEDGER} file_gone={not os.path.exists(order)}")
    print(f"[graph] fsm={kernel.state()}")
    timeline = kernel.replay_session(kernel.session_id())
    print(
        f"[graph] replay: {timeline['total_steps']} steps, "
        f"{timeline['compensations_executed']} compensations, "
        f"final={timeline['final_status']}"
    )


if __name__ == "__main__":
    main()