aiyoplane-langgraph

AEAP Composition Boundary Adapter for LangGraph.

A thin wrapper that attaches the AEAP Runtime Decision Point (RDP) to LangGraph's node-invocation surface. Consequential graph transitions produce independently verifiable Execution Receipts. The Composition Boundary sits at the graph node — the exact point where state transitions into a consequential action.

pip install aiyoplane-langgraph

Why LangGraph (vs. LangChain)

LangChain's Composition Boundary is the Tool. LangGraph's Composition Boundary is the node — a function (state) -> state_update. If you are building long-running agents, stateful workflows, multi-step execution with branches, or human-in-the-loop checkpoints, LangGraph is where the Runtime Decision Point most naturally lives. Every node that crosses a consequential boundary can produce an independently verifiable authorization decision.

See the AEAP specification §9 for the Composition Boundary primitive this adapter realizes.

Install

pip install aiyoplane-langgraph
# Peer dependencies installed automatically:
#   aiyoplane-mcp-authz >= 1.0.1
#   langgraph >= 0.2, < 0.3

Minimal usage

As a decorator

from langgraph.graph import StateGraph
from aiyoplane_mcp_authz import create_aiyo_mcp_authz, create_local_rdp
from aiyoplane_langgraph import aiyo_gate

policy = {
    "rules": [
        {"action_type": "deploy", "action_target": "production", "effect": "escalate"},
        {"action_type": "deploy", "action_target": "staging", "effect": "allow"},
        {"action_type": "deploy", "effect": "block"},
    ]
}

authz = create_aiyo_mcp_authz(
    rdp=create_local_rdp(policy=policy),
    tool_config={
        "deploy_node": {"aiyo_gated": True, "action": {"type": "deploy"}},
    },
)

@aiyo_gate(
    authz=authz,
    tool_name="deploy_node",
    intent_builder=lambda state: {
        "type": "deploy",
        "target": state["env"],
        "artifact": state["artifact_hash"],
    },
)
def deploy_node(state):
    # Runs only if the RDP returned ALLOW.
    # state["aeap_receipt"] contains the verified Execution Receipt.
    perform_deployment(state["artifact_hash"], state["env"])
    return {"deployed": True, "receipt": state["aeap_receipt"]}

graph = StateGraph(MyState)
graph.add_node("deploy", deploy_node)
# ... add edges, compile, invoke ...

As a function wrapper

from aiyoplane_langgraph import wrap_node

def deploy_node(state):
    perform_deployment(state["artifact_hash"], state["env"])
    return {"deployed": True}

guarded = wrap_node(
    deploy_node,
    authz=authz,
    tool_name="deploy_node",
    intent_builder=lambda state: {
        "type": "deploy",
        "target": state["env"],
        "artifact": state["artifact_hash"],
    },
)

graph.add_node("deploy", guarded)

Both forms produce the same result. Use the decorator when you own the node definition; use wrap_node when wrapping a node imported from elsewhere.

Composition Boundary — what the adapter actually does

For each wrapped node invocation:

1. Intent construction. The adapter calls intent_builder(state) to produce an AEAP Intent payload. See AEAP §3.
2. RDP evaluation. The Intent is passed to the AiyoAuthz middleware from aiyoplane-mcp-authz, which calls the RDP (local or hosted).
3. Verdict composition.
   • ALLOW → the inner node runs with an enriched state containing the Execution Receipt under the aeap_receipt key (overridable via attach_receipt_key=...).
   • ESCALATE → EscalationRequired is raised. Catch it in a conditional edge to route the graph into an approval branch.
   • BLOCK → BlockedByPolicy is raised. The node never executes.
4. Fail-closed default. Any ambiguous condition — missing intent_builder, non-dict Intent, RDP unreachable, misconfigured tool_config — results in BlockedByPolicy. See AEAP §2.2 invariant 5.

Routing escalations in a graph

The idiomatic LangGraph pattern is to route escalations through a conditional edge:

from aiyoplane_langgraph import EscalationRequired

def escalation_router(state):
    # If the previous node raised EscalationRequired, state contains
    # an "escalation" key (set by your own error handler).
    if state.get("escalation"):
        return "human_approval"
    return "continue"

graph.add_conditional_edges("deploy", escalation_router, {
    "human_approval": "approval_node",
    "continue": "notify_success",
})

The adapter raises EscalationRequired as a plain Python exception; wrap it in your own try/except in a wrapper node, or use LangGraph's error-handling features to catch and reroute.

Receipts attached to state

On ALLOW, the verified Execution Receipt is attached to state before the inner node runs:

@aiyo_gate(authz=authz, tool_name="deploy_node", intent_builder=...)
def deploy_node(state):
    receipt = state["aeap_receipt"]  # Ed25519-signed, verifiable offline
    perform_deployment(...)
    return {"deployed": True, "receipt": receipt}

The receipt is portable. Downstream nodes, downstream workflows, external systems, and audit pipelines can verify it offline against Aiyo's published JWKS using aiyoplane-verify — no coordination with the issuing RDP required at verification time.

Local vs. hosted RDP

from aiyoplane_mcp_authz import create_local_rdp, create_hosted_rdp

# Development / self-hosted:
authz = create_aiyo_mcp_authz(rdp=create_local_rdp(policy=policy), tool_config={...})

# Production / hosted:
authz = create_aiyo_mcp_authz(
    rdp=create_hosted_rdp(api_key="aiyo_live_..."),
    tool_config={...},
)

API reference

aiyo_gate(*, authz, tool_name, intent_builder, attach_receipt_key="aeap_receipt")

Decorator. Returns a decorator that wraps a node function.

wrap_node(node, *, authz, tool_name, intent_builder, attach_receipt_key="aeap_receipt")

Function form of aiyo_gate. Returns the wrapped node.

• node — the original node function (sync or async)
• authz — an AiyoAuthz instance
• tool_name — key in authz.tool_config
• intent_builder — (state: dict) -> dict
• attach_receipt_key — state key under which the receipt is attached on ALLOW

Exceptions

• AiyoLangGraphError — base class
• AdapterConfigError — adapter configuration is invalid
• BlockedByPolicy — RDP returned BLOCK; the node did not execute
• EscalationRequired — RDP returned ESCALATE; the node is held pending resolution

License

Apache License 2.0. See LICENSE for the full text.

Links

• AEAP specification: https://github.com/aiyoplane/aeap
• Underlying implementation: aiyoplane-mcp-authz
• Companion adapter: aiyoplane-langchain (LangChain Tool composition)
• Node sibling: @aiyoplane/langgraph on npm
• Trust surface: https://aiyoplane.com/trust
• Issues: https://github.com/aiyoplane/langgraph-python/issues

LangGraph is a trademark of LangChain, Inc. This package is an independent AEAP Composition Boundary Adapter and is not affiliated with or endorsed by LangChain, Inc.

Verify First. Execute Second.