Framework integrations

from langchain_core.messages import ToolMessage
from plyra_guard import ActionGuard, ActionIntent
from plyra_guard.exceptions import ExecutionBlockedError, ActionEscalatedError

guard = ActionGuard.from_config("guard_config.yaml")
TOOLS = {
    "read_file": read_file_tool,
    "delete_file": delete_file_tool,
}

def guarded_tool_node(state):
    messages = []
    for tool_call in state["messages"][-1].tool_calls:
        tool_name = tool_call["name"]
        args = tool_call["args"]

        # Build an ActionIntent for evaluation
        intent = ActionIntent(
            action_type=tool_name,
            tool_name=tool_name,
            parameters=args,
            agent_id="langgraph-agent",
        )

        try:
            result = guard.evaluate(intent)
            if result.verdict.is_blocking():
                content = f"[BLOCKED] {result.reason}"
            else:
                content = TOOLS[tool_name].invoke(args)
        except ExecutionBlockedError as e:
            content = f"[BLOCKED] {e.what_happened}"
        except ActionEscalatedError as e:
            content = f"[ESCALATED — awaiting approval] {e.reason}"
        except Exception as e:
            content = f"[ERROR] {e}"

        messages.append(
            ToolMessage(content=str(content), tool_call_id=tool_call["id"])
        )
    return {"messages": messages}

from plyra_guard import ActionGuard

guard = ActionGuard.from_config("guard_config.yaml")
safe_tools = guard.wrap(tools)

agent = create_react_agent(llm, safe_tools, prompt)

import autogen
from plyra_guard import ActionGuard

guard = ActionGuard.from_config("guard_config.yaml")
safe_tools = guard.wrap([read_file, delete_file])

user_proxy = autogen.UserProxyAgent("user_proxy", human_input_mode="NEVER")
for tool in safe_tools:
    user_proxy.register_function(function_map={tool.__name__: tool})

from plyra_guard import ActionGuard

guard = ActionGuard.from_config("guard_config.yaml")
safe_tools = guard.wrap([write_report])
agent = Agent(role="Analyst", tools=safe_tools, ...)

safe_tools = guard.wrap(tool_definitions)

from plyra_guard.exceptions import ExecutionBlockedError, ActionEscalatedError

try:
    my_function("arg")
except ExecutionBlockedError as e:
    # Action was blocked by policy
    print(e.what_happened, e.how_to_fix)
except ActionEscalatedError as e:
    # Action requires human approval before proceeding
    print(f"Escalated to: {e.escalate_to}")