iop-framework 0.1.0

Intent-Oriented Programming: FastAPI middleware + formal multi-agent contracts with SEPA and Assume-Guarantee guarantees

IOP Framework

Intent-Oriented Programming for FastAPI — formal multi-agent contracts with stochastic guarantees.

---

What is it?

IOP Framework converts any FastAPI app into an Intent-Oriented Programming system by adding three things:

1. @capability decorator — declare token preconditions/postconditions on your routes
2. IOPMiddleware — enforces contracts before every request (Semantic Firewall)
3. iop.core — formal engine: SEPA state machines, Assume-Guarantee contracts, Generative Sagas, MA-IRP protocol

Based on the research paper: IOP-Framework: From Semantic Gateway to a Multi-Agent Programming Paradigm with Stochastic EPA and Assume-Guarantee Contracts (arXiv:2604.25555).

---

Quick Start

Install

pip install iop-framework
# With LLM support (Generative Saga Engine):
pip install iop-framework[llm]

Convert a CRUD API to IOP in 3 lines

Before (plain FastAPI):

@app.post("/orders")
async def place_order(req: OrderRequest):
    ...

After (IOP-enabled):

from iop import capability, IOPMiddleware, register_app

app = FastAPI()
app.add_middleware(IOPMiddleware)          # ← line 1

@app.post("/orders")
@capability(                               # ← line 2
    pre=["USER_AUTHENTICATED", "PORTFOLIO_AUTHORIZED"],
    post=["ORDER_PLACED", "COLLATERAL_RESERVED"],
    description="Place a new trade order",
    drift=0.05,
)
async def place_order(req: OrderRequest):
    ...  # your code is unchanged

register_app(app)                          # ← line 3 (after all routes)

Now every call to POST /orders automatically:

• Returns 403 if USER_AUTHENTICATED or PORTFOLIO_AUTHORIZED tokens are missing
• Returns 503 if the stochastic drift d(κ) > 0.30
• Injects ORDER_PLACED and COLLATERAL_RESERVED into the session after success
• Adds X-IOP-Session, X-IOP-Capability, X-IOP-Drift, X-IOP-Tokens response headers

---

Full Trading API Example

from fastapi import FastAPI, Request
from iop import capability, IOPMiddleware, register_app, RiskLevel, CapabilityType

app = FastAPI(title="Trading API (IOP-enabled)")
app.add_middleware(IOPMiddleware, audit_log=True)

# ── Agent A1: Trader ────────────────────────────────
@app.post("/orders")
@capability(
    pre=["USER_AUTHENTICATED", "PORTFOLIO_AUTHORIZED"],
    post=["ORDER_PLACED", "COLLATERAL_RESERVED"],
    drift=0.05,
    description="Place a trade order and reserve collateral",
)
async def place_order(req: OrderRequest, request: Request):
    return {"order_id": "ord-001", "status": "placed"}

# ── Agent A2: Risk Manager ──────────────────────────
@app.post("/risk/evaluate")
@capability(
    pre=["ORDER_PLACED", "COLLATERAL_RESERVED"],
    post=["RISK_SCORE_COMPUTED", "RISK_EVALUATED"],
    drift=0.15,   # higher: LLM-based risk model
    description="Evaluate portfolio risk",
)
async def evaluate_risk(req: RiskRequest):
    return {"risk_score": 0.35, "risk_band": "MEDIUM"}

# ── Agent A3: Compliance Officer ────────────────────
@app.post("/compliance/verify")
@capability(
    pre=["RISK_SCORE_COMPUTED", "RISK_EVALUATED"],
    post=["MIFID_COMPLIANT", "TRADE_AUTHORIZED"],
    drift=0.08,
    description="MiFID II Article 25 suitability assessment",
)
async def verify_compliance(req: ComplianceRequest):
    return {"verdict": "AUTHORIZED", "mifid_ref": "Art.25(2014/65/EU)"}

# ── Compensating capability (Saga) ──────────────────
@app.post("/orders/{order_id}/cancel")
@capability(
    pre=["ORDER_PLACED"],
    post=["ORDER_CANCELLED"],
    revoke=["ORDER_PLACED", "COLLATERAL_RESERVED"],
    tau=CapabilityType.COMPENSATING,
    description="Cancel order and release collateral",
    drift=0.02,
)
async def cancel_order(order_id: str):
    return {"status": "cancelled", "collateral_released": True}

register_app(app)

Seed tokens via headers

# Authenticated call — passes through
curl -X POST http://localhost:8000/orders \
     -H "X-IOP-Seed-Tokens: USER_AUTHENTICATED,PORTFOLIO_AUTHORIZED" \
     -H "Content-Type: application/json" \
     -d '{"symbol":"AAPL","qty":100,"side":"buy"}'

# Missing token — 403
curl -X POST http://localhost:8000/orders \
     -H "Content-Type: application/json" \
     -d '{"symbol":"AAPL","qty":100,"side":"buy"}'
# → {"error":"precondition_not_met","message":"Missing tokens: ['USER_AUTHENTICATED']"}

---

Formal Engine (iop.core)

The formal engine powers the theoretical guarantees from the paper:

import asyncio
from iop.core import MAIRPOrchestrator

async def main():
    orch = MAIRPOrchestrator(rng_seed=42)

    # Clean flow
    result = await orch.execute(
        intent="BUY 100 AAPL",
        params={"ticker": "AAPL", "quantity": 100}
    )
    print(f"Success: {result.success}, Phases: {result.phases_completed}")

    # Authorization drift injection (for testing)
    result = await orch.execute(
        intent="BUY 500 TSLA",
        params={"ticker": "TSLA", "quantity": 500},
        inject_failure="drift",   # Risk agent skips collateral
    )
    print(f"Violation: {result.contract_violations}")
    print(f"Saga invoked: {result.saga_result is not None}")

asyncio.run(main())

Fuzzer / Validation

import asyncio
from iop.validation.fuzzer import MultiAgentFuzzer

async def main():
    fuzzer = MultiAgentFuzzer(seed=42)
    report = await fuzzer.run(n_iterations=1000, runs=10)
    print(report.summary())
    print(f"H1 Interception rate: {report.contract_interception_rate:.1%}")
    print(f"H3 Saga SRR: {report.saga_srr:.1%}")

asyncio.run(main())

---

Architecture

iop/                        ← FastAPI integration layer
├── capability.py           @capability decorator
├── middleware.py           IOPMiddleware (Semantic Firewall)
├── context.py              IopSession (per-request state)
└── types.py                Token, TokenContext, IntentContract, ...

iop/core/                   ← Formal engine (from paper PoC)
├── sepa.py                 StochasticEPA  M = (S̃, s̃₀, K, P)
├── contracts.py            ContractDefinition, ContractLedger, ContractVerifier
├── agents.py               BaseAgent (abstract stochastic agent)
├── financial_agents.py     Trader, RiskManagementAgent, ComplianceAgent
├── sagas.py                GenerativeSagaEngine, LLMAuditor
└── orchestrator.py         MAIRPOrchestrator (6-phase MA-IRP)

iop/validation/
├── fuzzer.py               MultiAgentFuzzer (SEPA-guided)
└── baseline.py             LangGraphSimulator (no-contract baseline)

---

Theoretical Guarantees

Theorem 4 (SEPA Soundness)

If capability κ satisfies EPA Target Validity:

P(κ_next ∈ enabled_R(s̃_next)) ≥ 1 − d(s̃, κ)

DRIFT_THRESHOLD = 0.30 is derived from this bound, not tuned empirically.

Theorem 6 (AG Compositionality)

For a chain A₁ → A₂ → A₃ with drift rates d₁, d₂, d₃:

P(Ψ₃ | Φ₁) ≥ P(Ψ₁ | Φ₁) · (1−d₁)(1−d₂)(1−d₃)

For the trading chain (d₁=0.05, d₂=0.15, d₃=0.08): ≥ 0.744

Empirical Results (20,000 flows, 20 runs)

Hypothesis	Result	Verdict
H1: Contract interception rate > 80%	91.39%	✓ SUPPORTED
H2: MTTD improvement (BOLA)	5.65×	✓ SUPPORTED
H2: MTTD improvement (CTX poisoning)	4.81×	✓ SUPPORTED
H3: Saga recovery rate	100%	✓ SUPPORTED
H4: FPR on benign flows	0.00% [0, 0.018%]	✓ SUPPORTED

---

Token Seeding

In production, tokens are typically injected by your auth middleware:

@app.middleware("http")
async def auth_middleware(request: Request, call_next):
    token = request.headers.get("Authorization", "")
    if verify_jwt(token):
        # Seed IOP tokens from JWT claims
        request.headers.__dict__["_list"].append(
            (b"x-iop-seed-tokens",
             b"USER_AUTHENTICATED,PORTFOLIO_AUTHORIZED")
        )
    return await call_next(request)

Or via initial_tokens in the middleware constructor for a fixed global seed:

app.add_middleware(IOPMiddleware, initial_tokens=["SERVICE_ACCOUNT"])

---

Contributing

git clone https://github.com/PeyranoDev/iop-framework
cd iop-framework
python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
pytest tests/ -v

---

Citation

@misc{peyrano2025iop,
  author    = {Peyrano, Ignacio},
  title     = {From {CRUD} to Autonomous Agents: Formal Validation and
               Zero-Trust Security for Semantic Gateways in {AI}-Native
               Enterprise Systems},
  year      = {2025},
  eprint    = {2604.25555},
  archivePrefix = {arXiv},
}

---

License

MIT © Ignacio Peyrano — Universidad Austral, Rosario, Argentina