LangGraph

LangGraph

Add a Xybern gate as a dedicated node in your LangGraph state graph. The gate runs before any execution node and sets an xybern_approved flag in state, the execution node checks this flag before proceeding.

# langgraph_xybern.py
import requests
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage

XYBERN_API_KEY = "xb_your_api_key"

class AgentState(TypedDict):
    messages: list
    next_action: str
    xybern_approved: bool

def xybern_gate(state: AgentState) -> AgentState:
    """LangGraph node: authorise the next action before execution."""
    next_action = state.get("next_action", "")
    if not next_action:
        return {**state, "xybern_approved": True}

    resp = requests.post(
        "https://www.xybern.com/api/v1/enforce/intercept",
        headers={"X-API-Key": XYBERN_API_KEY},
        json={
            "agent_id": "langgraph_agent_001",
            "action": {"type": "tool_call", "tool": next_action, "parameters": {}},
            "context": {"framework": "langgraph", "state_step": len(state["messages"])}
        }
    ).json()

    return {**state, "xybern_approved": resp["decision"] == "ALLOW"}

def execute_action(state: AgentState) -> AgentState:
    if not state["xybern_approved"]:
        msg = AIMessage(content="Action denied by Xybern authorisation policy.")
        return {**state, "messages": state["messages"] + [msg]}
    msg = AIMessage(content=f"Executing {state['next_action']}...")
    return {**state, "messages": state["messages"] + [msg], "next_action": ""}

graph = StateGraph(AgentState)
graph.add_node("xybern_gate", xybern_gate)
graph.add_node("execute", execute_action)
graph.set_entry_point("xybern_gate")
graph.add_edge("xybern_gate", "execute")
graph.add_edge("execute", END)

app = graph.compile()
result = app.invoke({
    "messages": [HumanMessage(content="Run report")],
    "next_action": "generate_report",
    "xybern_approved": False
})