diff --git a/.github/workflows/py-sdk-build-publish.yml b/.github/workflows/py-sdk-build-publish.yml index b733106..e0954b9 100644 --- a/.github/workflows/py-sdk-build-publish.yml +++ b/.github/workflows/py-sdk-build-publish.yml @@ -119,7 +119,7 @@ jobs: publish_test: name: Publish to Test PyPI - if: ${{ github.ref_type == 'branch' && github.ref_name != 'main' }} + if: ${{ github.event_name == 'workflow_dispatch'}} needs: [build_wheels, build_sdist] runs-on: ubuntu-latest #environment: pypi diff --git a/Cargo.toml b/Cargo.toml index 5856a63..e716667 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,7 +8,7 @@ members = [ resolver = "2" [workspace.package] -version = "0.4.0" +version = "0.5.0" edition = "2021" license = "Apache-2.0" authors = ["Amit Saxena"] diff --git a/README.md b/README.md index 13e099c..d7b593d 100644 --- a/README.md +++ b/README.md @@ -1,216 +1,234 @@ # Actra -Deterministic admission control engine for state-changing operations in automated and agentic systems. +[![PyPI version](https://img.shields.io/pypi/v/actra.svg)](https://pypi.org/project/actra/) +[![PyPI downloads](https://img.shields.io/pypi/dm/actra)](https://pypi.org/project/actra/) -Actra compiles declarative policy definitions into a validated, -immutable intermediate representation (IR) that can be evaluated -efficiently against runtime inputs. +**Action Admission Control for Automated Systems** -It is designed for systems that require explicit, reproducible control -over mutations - including API gateways, background workers, -agentic runtimes and automated service workflows. +Deterministic policy engine that decides whether automated actions are **allowed before they execute**. ---- - -## Why Actra? +Actra prevents unsafe operations in: -Modern systems often mix: +* AI agents +* APIs +* automation systems +* background workers +* workflows -- Business logic -- Authorization logic -- Governance rules -- Operational constraints +Instead of embedding control logic in application code, Actra evaluates **external policies** before state-changing actions run. -This leads to implicit, scattered mutation controls. +--- -Actra separates these concerns: +![MCP Demo](doc/mcp-demo.gif) -1. **Schema** defines allowed structure. -2. **Policy** defines decision logic. -3. **Governance** defines organizational constraints. -4. **Compiler** validates everything upfront. -5. **Engine** performs deterministic evaluation at runtime. +Agent attempted to call an MCP tool. -All semantic validation happens at compile time. -Runtime evaluation performs no structural checks. +Actra evaluated policy and **blocked the unsafe operation before execution** --- -## Core Design Principles - -- Deterministic evaluation -- Strict compile-time validation -- First-match-wins rule semantics -- Default allow (unless explicitly blocked) -- No unsafe Rust -- Runtime-agnostic core +## Why Actra? ---- +Modern systems increasingly perform actions automatically: -## Applicable Systems +* AI agents calling tools +* workflow automation +* API integrations +* background jobs -Actra is designed for systems that perform controlled mutations, including: +These systems can trigger **powerful state-changing operations**, such as: -- Agentic AI runtimes -- Tool-executing LLM systems -- Automation pipelines -- Workflow engines -- API gateways -- State mutation services +* issuing refunds +* deleting resources +* sending payments +* modifying infrastructure -It is particularly relevant for: -- Agent frameworks such as OpenClaw -- Systems implementing the Model Context Protocol (MCP) -- Automated remediation systems -- Infrastructure orchestration engines +Today these controls often live inside application code: -In these environments, LLMs or automated processes may attempt -state-changing operations. Actra provides a deterministic, -compile-time validated control layer in front of those mutations. +```python +if amount > 1000: + raise Exception("Refund too large") +``` -It does not replace orchestration logic or workflow execution. -It acts as an explicit mutation admission boundary between -decision-making systems and state-changing operations. +This creates problems: -Typical embedding points include: +* rules duplicated across services +* difficult to audit behavior +* policy changes require redeploys +* automation becomes risky -- Agent execution runtimes -- Tool invocation layers -- MCP-based tool servers -- Background automation workers -- API request pipelines +Actra moves these decisions into **deterministic external policies evaluated before the action executes**. --- -## Example (Node.js) - -```js -const { Actra } = require('./index') - -const schema = ` -version: 1 -actions: - delete_user: - fields: - type: string - user_id: string -actor: - fields: - role: string -snapshot: - fields: - account_tier: string -` - -const policy = ` -version: 1 +## 20-Second Example + +```python +@actra.admit() +def refund(amount): + ... +``` + +The rule lives in policy: + +```yaml rules: - - id: block_non_admin_delete - scope: - action: delete_user + - id: block_large_refund when: subject: - domain: actor - field: role - operator: equals + domain: action + field: amount + operator: greater_than value: - literal: "admin" + literal: 1000 effect: block -` +``` -const actra = new Actra(schema, policy) +```markdown +Result: + +refund(200) > allowed +refund(1500) > blocked by policy +``` -const result = actra.evaluate({ - action: { type: "delete_user", user_id: "123" }, - actor: { role: "user" }, - snapshot: {} -}) +Actra evaluates the policy **before the function executes** and blocks refunds greater than 1000. -console.log(result) -// { effect: "allow", matched_rule: "" } +--- + +## Installation + +```bash +pip install actra ``` -## Example (Python) +See the **examples/** directory for quick start examples. -```python -from actra import Actra - -schema = """ -version: 1 -actions: - update_account: - fields: - type: string - account_id: string -actor: - fields: - role: string -snapshot: - fields: - region: string -""" - -policy = """ -version: 1 -rules: - - id: block_non_admin - scope: - action: update_account - when: - subject: - domain: actor - field: role - operator: equals - value: - literal: "admin" - effect: block -""" +--- -actra = Actra(schema, policy) +## Architecture -result = actra.evaluate({ - "action": {"type": "update_account", "account_id": "A1"}, - "actor": {"role": "user"}, - "snapshot": {} -}) +Actra evaluates policies **before operations execute**. -print(result) -``` +```mermaid +flowchart LR -## Governance Layer +A[Application / Agent / API] --> B[Action Request] -Actra includes an optional governance DSL that allows higher-order constraints on policies: +B --> C[Actra Admission Control] -- Require certain rules to exist -- Restrict allowed fields -- Limit rule counts -- Forbid specific patterns +C --> D[Schema] +C --> E[Policies] +C --> F[Governance optional] +C --> G[Runtime Context] -Governance validation runs before compilation and can reject otherwise valid policies. +G --> G1[Actor] +G --> G2[Action] +G --> G3[Snapshot] -Architecture +C --> H{Decision} -```mermaid -flowchart TD - A[Schema - YAML] - B[Policy AST - YAML] - C[Governance Validation - Optional] - D[Compiler] - E[Compiled IR] - F[Evaluation Engine - Runtime] - - A --> B --> C --> D --> E --> F +H -->|Allow| I[Execute Operation] +H -->|Block| J[Operation Prevented] ``` -The evaluation engine operates only on validated IR. -No dynamic validation occurs at runtime. +--- + +## Example Use Cases + +Actra can control many automated operations. -## Status +### AI Agents -Actra is currently in early foundation stage (v0.1.x). +* restrict tool execution +* prevent critical infrastructure changes +* enforce safety policies -Core architecture is stable. -Public API may evolve as the DSL matures. +### APIs + +* block large refunds +* prevent destructive operations +* enforce safety checks + +### Automation + +* enforce workflow rules +* restrict financial operations +* require approval thresholds + +### Infrastructure + +* prevent destructive changes +* enforce safe deployment policies + +--- + +## SDKs + +Actra supports multiple runtimes. + +| Runtime | Status | +| ------- | ------------ | +| Python | Available | +| Node.js | WIP | +| Rust | Core runtime | +| WASM | Planned | +| Go | Planned | + +--- + +## Actra vs OPA vs Cedar + +| Feature | Actra | OPA | Cedar | +| ----------------- | -------------------------------- | ------------------------------ | ----------------------------- | +| Primary purpose | Admission control for operations | General policy engine | Authorization policy language | +| Evaluation timing | **Before executing actions** | Usually request-time decisions | Authorization decisions | +| Integration model | Function / action enforcement | API / sidecar / middleware | Service authorization | +| Policy style | Structured YAML rules | Rego language | Cedar language | +| Determinism focus | Strong | Moderate | Strong | +| Target systems | Agents, automation, APIs | Infrastructure, Kubernetes | Application authorization | +| Typical use case | Block unsafe operations | Policy enforcement in infra | Access control | + +### Positioning + +Actra focuses on **controlling actions before they execute**, especially in automated or agent-driven systems. + +OPA and Cedar focus primarily on **authorization decisions**, such as: + +* “Can user X access resource Y?” + +Actra focuses on **admission control for mutations**, such as: + +* Should this refund execute? +* Should an agent run this tool? +* Should this workflow step proceed? + +### Example Scenarios + +| Scenario | Best Tool | +| ------------------------------------------------ | --------- | +| Can a user access a document? | Cedar | +| Can a service access an API? | OPA | +| Should an automated system execute an operation? | Actra | + + +--- + +## Documentation + +Full documentation coming soon. + +Refer to the **examples** folder for detailed usage examples. + +Planned documentation sections: + +* policy language +* MCP integration +* agent safety +* runtime architecture +* advanced policy patterns + +--- ## License -This project is licensed under the Apache License 2.0 - see the LICENSE file for details. \ No newline at end of file + +Apache 2.0 diff --git a/doc/mcp-demo.gif b/doc/mcp-demo.gif new file mode 100644 index 0000000..d073aad Binary files /dev/null and b/doc/mcp-demo.gif differ diff --git a/examples/python/advance/ai_agent_tool_guardrail.py b/examples/python/advance/ai_agent_tool_guardrail.py new file mode 100644 index 0000000..4897b9c --- /dev/null +++ b/examples/python/advance/ai_agent_tool_guardrail.py @@ -0,0 +1,193 @@ +""" +Actra Advanced Example +AI Agent Tool Guardrails + +This example demonstrates how Actra can enforce safety policies +for AI agents executing external tools + +Large language model agents often have access to powerful tools such as: + +• sending emails +• issuing refunds +• deleting resources +• executing infrastructure operations + +Without guardrails, an AI agent may accidentally perform destructive +or high-impact operations + +Actra acts as a deterministic admission control layer between the +AI agent and the underlying tools + +The policy in this example enforces: + +• AI agents cannot issue refunds above a safe limit +• Only human supervisors can delete users +• High-value refunds require supervisor role + +This pattern allows AI systems to remain autonomous while still +operating within strict operational boundaries +""" + +from actra import Actra, ActraRuntime, ActraPolicyError + + +# ------------------------------------------------------------ +# Schema +# ------------------------------------------------------------ + +schema_yaml = """ +version: 1 + +actions: + + issue_refund: + fields: + amount: number + currency: string + + delete_user: + fields: + user_id: string + +actor: + fields: + role: string + agent_type: string + +snapshot: + fields: + daily_refund_total: number +""" + +# ------------------------------------------------------------ +# Policy +# ------------------------------------------------------------ + +policy_yaml = """ +version: 1 + +rules: + + # AI agents cannot issue refunds above $500 + - id: ai_refund_limit + scope: + action: issue_refund + when: + all: + - subject: + domain: actor + field: agent_type + operator: equals + value: + literal: "ai" + + - subject: + domain: action + field: amount + operator: greater_than + value: + literal: 500 + effect: block + + + # Only supervisors can delete users + - id: restrict_user_deletion + scope: + action: delete_user + when: + subject: + domain: actor + field: role + operator: not_equals + value: + literal: "supervisor" + effect: block +""" + +# ------------------------------------------------------------ +# Compile policy +# ------------------------------------------------------------ + +policy = Actra.from_strings(schema_yaml, policy_yaml) +runtime = ActraRuntime(policy) + +# ------------------------------------------------------------ +# Runtime Context +# ------------------------------------------------------------ + +class AgentContext: + def __init__(self, role, agent_type): + self.role = role + self.agent_type = agent_type + + +runtime.set_actor_resolver( + lambda ctx: { + "role": ctx.role, + "agent_type": ctx.agent_type + } +) + +runtime.set_snapshot_resolver( + lambda ctx: { + "daily_refund_total": 1200 + } +) + +# ------------------------------------------------------------ +# AI Agent Tools +# ------------------------------------------------------------ + +@runtime.admit(action_type="issue_refund") +def refund(amount, currency, ctx=None): + print(f"Issuing refund: {amount} {currency}") + + +@runtime.admit(action_type="delete_user") +def delete_user(user_id, ctx=None): + print(f"Deleting user {user_id}") + +# ------------------------------------------------------------ +# Contexts +# ------------------------------------------------------------ + +ai_agent = AgentContext(role="assistant", agent_type="ai") +supervisor = AgentContext(role="supervisor", agent_type="human") + +# ------------------------------------------------------------ +# Allowed operation +# ------------------------------------------------------------ + +refund(200, "USD", ctx=ai_agent) + + +# ------------------------------------------------------------ +# Blocked AI refund +# ------------------------------------------------------------ + +try: + refund(1000, "USD", ctx=ai_agent) +except ActraPolicyError: + print("AI agent cannot issue large refunds") + +# ------------------------------------------------------------ +# Blocked user deletion +# ------------------------------------------------------------ + +try: + delete_user("user_123", ctx=ai_agent) +except ActraPolicyError: + print("Only supervisors can delete users") + +# ------------------------------------------------------------ +# Debug a decision +# ------------------------------------------------------------ + +print("\nExplain refund decision") +runtime.explain_call( + refund, + action_type="issue_refund", + amount=800, + currency="USD", + ctx=ai_agent +) diff --git a/examples/python/advance/ai_high_risk_action_approval.py b/examples/python/advance/ai_high_risk_action_approval.py new file mode 100644 index 0000000..3b516f7 --- /dev/null +++ b/examples/python/advance/ai_high_risk_action_approval.py @@ -0,0 +1,234 @@ +""" +Actra Advanced Example +AI High-Risk Action Approval + +This example demonstrates how Actra can enforce approval workflows +for high-risk operations triggered by automation or AI agents + +Certain operations should not be immediately blocked or allowed +Instead, they should require human review before execution + +Actra supports this pattern using the `require_approval` effect + +Scenario +-------- + +An AI automation system manages cloud infrastructure. It can perform +routine operations autonomously, but high-risk operations require +human approval + +The policy enforces the following rules: + +• AI agents scaling services beyond safe limits require approval +• Production deployments require operator privileges +• AI agents cannot delete production services + +This example shows how Actra can act as a deterministic control +layer for autonomous systems while still allowing human oversight +for critical operations. +""" + +from actra import Actra, ActraRuntime, ActraPolicyError + + +# ------------------------------------------------------------ +# Schema +# ------------------------------------------------------------ + +schema_yaml = """ +version: 1 + +actions: + + scale_service: + fields: + service: string + replicas: number + environment: string + + deploy_service: + fields: + service: string + version: string + environment: string + + delete_service: + fields: + service: string + environment: string + +actor: + fields: + role: string + agent_type: string + +snapshot: + fields: + cluster_capacity: number +""" + +# ------------------------------------------------------------ +# Policy +# ------------------------------------------------------------ + +policy_yaml = """ +version: 1 + +rules: + + # Large scaling operations require approval + - id: large_scale_requires_approval + scope: + action: scale_service + when: + subject: + domain: action + field: replicas + operator: greater_than + value: + literal: 15 + effect: require_approval + + + # Production deployments restricted + - id: prod_deploy_restricted + scope: + action: deploy_service + when: + all: + - subject: + domain: action + field: environment + operator: equals + value: + literal: "prod" + + - subject: + domain: actor + field: role + operator: not_equals + value: + literal: "operator" + effect: block + + + # AI agents cannot delete production services + - id: block_ai_delete_prod + scope: + action: delete_service + when: + all: + - subject: + domain: actor + field: agent_type + operator: equals + value: + literal: "ai" + + - subject: + domain: action + field: environment + operator: equals + value: + literal: "prod" + effect: block +""" + +# ------------------------------------------------------------ +# Compile policy +# ------------------------------------------------------------ + +policy = Actra.from_strings(schema_yaml, policy_yaml) +runtime = ActraRuntime(policy) + +# ------------------------------------------------------------ +# Context +# ------------------------------------------------------------ + +class RequestContext: + def __init__(self, role, agent_type): + self.role = role + self.agent_type = agent_type + + +runtime.set_actor_resolver( + lambda ctx: { + "role": ctx.role, + "agent_type": ctx.agent_type + } +) + +runtime.set_snapshot_resolver( + lambda ctx: { + "cluster_capacity": 20 + } +) + +# ------------------------------------------------------------ +# Protected operations +# ------------------------------------------------------------ + +@runtime.admit(action_type="scale_service") +def scale(service, replicas, environment, ctx=None): + print(f"Scaling {service} to {replicas} replicas") + + +@runtime.admit(action_type="deploy_service") +def deploy(service, version, environment, ctx=None): + print(f"Deploying {service}:{version} to {environment}") + + +@runtime.admit(action_type="delete_service") +def delete(service, environment, ctx=None): + print(f"Deleting service {service}") + +# ------------------------------------------------------------ +# Contexts +# ------------------------------------------------------------ + +ai_agent = RequestContext(role="automation", agent_type="ai") +operator = RequestContext(role="operator", agent_type="human") + + +# ------------------------------------------------------------ +# Allowed operation +# ------------------------------------------------------------ + +scale("search-api", replicas=5, environment="staging", ctx=ai_agent) + + +# ------------------------------------------------------------ +# Operation requiring approval +# ------------------------------------------------------------ + +try: + scale("search-api", replicas=20, environment="staging", ctx=ai_agent) +except ActraPolicyError as e: + if e.decision.get("effect") == "require_approval": + print("Scaling requires human approval") + + +# ------------------------------------------------------------ +# Blocked destructive action +# ------------------------------------------------------------ + +try: + delete("search-api", environment="prod", ctx=ai_agent) +except ActraPolicyError: + print("AI cannot delete production services") + + +# ------------------------------------------------------------ +# Debug policy decision +# ------------------------------------------------------------ + +print("\nExplain decision for scaling") + +runtime.explain_call( + scale, + action_type="scale_service", + service="search-api", + replicas=20, + environment="staging", + ctx=operator +) \ No newline at end of file diff --git a/examples/python/advance/ai_infrastructure_guardrails.py.py b/examples/python/advance/ai_infrastructure_guardrails.py.py new file mode 100644 index 0000000..cc7a631 --- /dev/null +++ b/examples/python/advance/ai_infrastructure_guardrails.py.py @@ -0,0 +1,303 @@ +""" +Actra Advanced Example +AI Infrastructure Guardrails + +This example demonstrates how Actra can enforce cloud infrastructure +operations performed by humans, automation systems, or AI agents + +The policy enforces safety controls for common platform operations such as: + +• Scaling services +• Deploying new versions +• Deleting services + +The example showcases several Actra capabilities: + +- Multiple action types defined in the schema +- Actor identity resolution (human vs AI agents) +- Snapshot-based policies using live system state +- Policy enforcement using the `@runtime.admit` decorator +- Safe blocking of destructive operations +- Debugging policy decisions using `runtime.explain_call` + +Scenario +-------- + +A platform team operates a service platform where operations may be +triggered by: + +- human operators +- automation pipelines +- AI agents managing infrastructure + +Actra ensures that: + +- AI agents cannot delete production services +- Production deployments require operator privileges +- Services cannot scale beyond cluster capacity + +This example illustrates how Actra acts as a deterministic +admission control layer for infrastructure automation. +""" + +from actra import Actra, ActraRuntime, ActraPolicyError + + +# ------------------------------------------------------------ +# Schema +# ------------------------------------------------------------ + +schema_yaml = """ +version: 1 + +actions: + + scale_service: + fields: + service: string + replicas: number + environment: string + + deploy_service: + fields: + service: string + version: string + environment: string + + delete_service: + fields: + service: string + environment: string + +actor: + fields: + role: string + agent_type: string + +snapshot: + fields: + cluster_capacity: number + production_freeze: boolean +""" + + +# ------------------------------------------------------------ +# Policy +# ------------------------------------------------------------ + +policy_yaml = """ +version: 1 + +rules: + + # AI agents cannot delete production services + - id: block_ai_delete_prod + scope: + action: delete_service + when: + all: + - subject: + domain: actor + field: agent_type + operator: equals + value: + literal: "ai" + + - subject: + domain: action + field: environment + operator: equals + value: + literal: "prod" + effect: block + + + # Production deployments require operator role + - id: prod_deploy_restricted + scope: + action: deploy_service + when: + all: + - subject: + domain: action + field: environment + operator: equals + value: + literal: "prod" + + - subject: + domain: actor + field: role + operator: not_equals + value: + literal: "operator" + effect: block + + + # Prevent scaling beyond cluster capacity + - id: cluster_capacity_limit + scope: + action: scale_service + when: + subject: + domain: action + field: replicas + operator: greater_than + value: + subject: + domain: snapshot + field: cluster_capacity + effect: block + + # Freeze Production Deployments + - id: freeze_production_deploys + scope: + action: deploy_service + when: + all: + - subject: + domain: snapshot + field: production_freeze + operator: equals + value: + literal: true + - subject: + domain: action + field: environment + operator: equals + value: + literal: "prod" + effect: block + +""" + + +# ------------------------------------------------------------ +# Compile policy +# ------------------------------------------------------------ + +policy = Actra.from_strings(schema_yaml, policy_yaml) +runtime = ActraRuntime(policy) + + +# ------------------------------------------------------------ +# Runtime context +# ------------------------------------------------------------ + +class RequestContext: + def __init__(self, role, agent_type, freeze = False): + self.role = role + self.agent_type = agent_type + self.freeze = freeze + + +runtime.set_actor_resolver( + lambda ctx: { + "role": ctx.role, + "agent_type": ctx.agent_type + } +) + +runtime.set_snapshot_resolver( + lambda ctx: { + "cluster_capacity": 10, + "production_freeze": ctx.freeze + } +) + + +# ------------------------------------------------------------ +# Protected operations +# ------------------------------------------------------------ + +@runtime.admit(action_type="scale_service") +def scale(service, replicas, environment, ctx=None): + print(f"Scaling {service} : {replicas} replicas") + + +@runtime.admit(action_type="deploy_service") +def deploy(service, version, environment, ctx=None): + print(f"Deploying {service}:{version} to {environment}") + + +@runtime.admit(action_type="delete_service") +def delete(service, environment, ctx=None): + print(f"Deleting service {service}") + + +# ------------------------------------------------------------ +# Contexts +# ------------------------------------------------------------ + +ai_agent = RequestContext(role="automation", agent_type="ai") + +operator = RequestContext(role="operator", agent_type="human") + +freeze_operator = RequestContext( + role="operator", + agent_type="human", + freeze=True +) + +# ------------------------------------------------------------ +# Allowed operation +# ------------------------------------------------------------ + +scale("search-api", replicas=5, environment="staging", ctx=ai_agent) + + +# ------------------------------------------------------------ +# Blocked AI destructive action +# ------------------------------------------------------------ + +try: + delete("search-api", environment="prod", ctx=ai_agent) +except ActraPolicyError: + print("AI cannot delete production services") + + +# ------------------------------------------------------------ +# Blocked deployment +# ------------------------------------------------------------ + +try: + deploy("search-api", version="2.1", environment="prod", ctx=ai_agent) +except ActraPolicyError: + print("Only operators can deploy to production") + + +print("\nProduction freeze example") + +try: + deploy( + "search-api", + version="2.5", + environment="prod", + ctx=freeze_operator + ) +except ActraPolicyError: + print("Deployment blocked due to production freeze") + + +# ------------------------------------------------------------ +# Policy Debugging with explain_call +# ------------------------------------------------------------ + +print("\nExplain call for scale service") +runtime.explain_call( + scale, + action_type="scale_service", + service="search-api", + replicas=20, + environment="staging", + ctx=operator +) + +print("\nExplain call for delete service") +runtime.explain_call( + delete, + action_type="delete_service", + service="lookup-customer", + environment="prod", + ctx=ai_agent +) diff --git a/examples/python/advance/ai_multi_step_safety.py b/examples/python/advance/ai_multi_step_safety.py new file mode 100644 index 0000000..5960ca9 --- /dev/null +++ b/examples/python/advance/ai_multi_step_safety.py @@ -0,0 +1,209 @@ +""" +Actra Advanced Example +AI Multi-Step Safety Guardrails + +This example demonstrates how Actra can enforce safety rules across +multiple operations performed by automation or AI agents + +Certain operations may be safe individually but dangerous when +combined together + +Scenario +-------- + +An AI agent manages infrastructure operations such as: + +• restarting services +• disabling monitoring +• deleting databases + +Deleting a database is normally restricted, but becomes even more +dangerous if monitoring has already been disabled + +Actra prevents this by using snapshot state to detect unsafe +multi-step sequences + +Policy Rules +------------ + +• AI agents cannot delete databases in production +• If monitoring is disabled, database deletion requires approval +""" + +from actra import Actra, ActraRuntime, ActraPolicyError + +# ------------------------------------------------------------ +# Schema +# ------------------------------------------------------------ + +schema_yaml = """ +version: 1 + +actions: + + disable_monitoring: + fields: + service: string + + delete_database: + fields: + database: string + environment: string + + restart_service: + fields: + service: string + +actor: + fields: + role: string + agent_type: string + +snapshot: + fields: + monitoring_disabled: boolean +""" + +# ------------------------------------------------------------ +# Policy +# ------------------------------------------------------------ + +policy_yaml = """ +version: 1 + +rules: + + # AI agents cannot delete production databases + - id: block_ai_db_delete + scope: + action: delete_database + when: + all: + - subject: + domain: actor + field: agent_type + operator: equals + value: + literal: "ai" + + - subject: + domain: action + field: environment + operator: equals + value: + literal: "prod" + effect: block + + + # If monitoring is disabled, deletion requires approval + - id: delete_requires_approval_when_monitoring_disabled + scope: + action: delete_database + when: + subject: + domain: snapshot + field: monitoring_disabled + operator: equals + value: + literal: true + effect: require_approval +""" + +# ------------------------------------------------------------ +# Compile policy +# ------------------------------------------------------------ + +policy = Actra.from_strings(schema_yaml, policy_yaml) +runtime = ActraRuntime(policy) + + +# ------------------------------------------------------------ +# Context +# ------------------------------------------------------------ + +class RequestContext: + def __init__(self, role, agent_type, monitoring_disabled=False): + self.role = role + self.agent_type = agent_type + self.monitoring_disabled = monitoring_disabled + + +runtime.set_actor_resolver( + lambda ctx: { + "role": ctx.role, + "agent_type": ctx.agent_type + } +) + +runtime.set_snapshot_resolver( + lambda ctx: { + "monitoring_disabled": ctx.monitoring_disabled + } +) + + +# ------------------------------------------------------------ +# Protected operations +# ------------------------------------------------------------ + +@runtime.admit() +def disable_monitoring(service, ctx=None): + print(f"Monitoring disabled for {service}") + + +@runtime.admit() +def delete_database(database, environment, ctx=None): + print(f"Database {database} deleted") + + +@runtime.admit() +def restart_service(service, ctx=None): + print(f"Restarting service {service}") + +# ------------------------------------------------------------ +# Contexts +# ------------------------------------------------------------ + +ai_agent = RequestContext( + role="automation", + agent_type="ai", + monitoring_disabled=True +) + +operator = RequestContext( + role="operator", + agent_type="human", + monitoring_disabled=True +) + +# ------------------------------------------------------------ +# Safe operation +# ------------------------------------------------------------ + +restart_service("search-api", ctx=ai_agent) + + +# ------------------------------------------------------------ +# Risky multi-step operation +# ------------------------------------------------------------ + +try: + delete_database("customer-db", environment="staging", ctx=ai_agent) +except ActraPolicyError as e: + if e.decision.get("effect") == "require_approval": + print("Database deletion requires approval because monitoring is disabled") + + +# ------------------------------------------------------------ +# Debug policy decision +# ------------------------------------------------------------ + +print("\nExplain decision") + +runtime.explain_call( + delete_database, + action_type="delete_database", + database="customer-db", + environment="staging", + ctx=operator +) \ No newline at end of file diff --git a/examples/python/helpers/action_evaluate_example.py b/examples/python/helpers/action_evaluate_example.py new file mode 100644 index 0000000..07228a6 --- /dev/null +++ b/examples/python/helpers/action_evaluate_example.py @@ -0,0 +1,110 @@ +""" +Actra Helper Example — action() + evaluate() + +This example demonstrates how to construct an action +programmatically using `runtime.action()` and evaluate +the policy decision directly + +This pattern is useful when there is no Python function +to decorate (for example APIs, agents, or event systems) +""" + +from actra import Actra +from actra.runtime import ActraRuntime + + +# ------------------------------------------------------------ +# 1. Schema definition +# ------------------------------------------------------------ + +schema_yaml = """ +version: 1 + +actions: + deploy: + fields: + service: string + env: string + +actor: + fields: + role: string + +snapshot: + fields: + maintenance_mode: boolean +""" + + +# ------------------------------------------------------------ +# 2. Policy definition +# ------------------------------------------------------------ + +policy_yaml = """ +version: 1 + +rules: + - id: block_prod_deploy + scope: + action: deploy + when: + subject: + domain: action + field: env + operator: equals + value: + literal: "prod" + effect: block +""" + + +# ------------------------------------------------------------ +# 3. Compile policy +# ------------------------------------------------------------ + +policy = Actra.from_strings(schema_yaml, policy_yaml) + + +# ------------------------------------------------------------ +# 4. Create runtime +# ------------------------------------------------------------ + +runtime = ActraRuntime(policy) + + +# ------------------------------------------------------------ +# 5. Register resolvers +# ------------------------------------------------------------ + +runtime.set_actor_resolver(lambda ctx: {"role": "devops"}) +runtime.set_snapshot_resolver(lambda ctx: {"maintenance_mode": False}) + + +# ------------------------------------------------------------ +# 6. Evaluate actions +# ------------------------------------------------------------ + +print("\n--- Allowed deployment ---") + +decision = runtime.evaluate( + runtime.action( + "deploy", + service="billing", + env="staging" + ) +) + +print("Decision:", decision) + + +print("\n--- Blocked deployment ---") + +decision = runtime.evaluate( + runtime.action( + "deploy", + service="billing", + env="prod" + ) +) + +print("Decision:", decision) \ No newline at end of file diff --git a/examples/python/helpers/allow_helper_example.py b/examples/python/helpers/allow_helper_example.py new file mode 100644 index 0000000..2539deb --- /dev/null +++ b/examples/python/helpers/allow_helper_example.py @@ -0,0 +1,88 @@ +""" +Actra Helper Example — allow() + +The allow() helper evaluates the policy and returns True +if the action is permitted + +This is useful when the application only needs a boolean +decision instead of the full policy result +""" + +from actra import Actra +from actra.runtime import ActraRuntime + + +# ------------------------------------------------------------ +# 1. Schema +# ------------------------------------------------------------ + +schema_yaml = """ +version: 1 + +actions: + scale_service: + fields: + service: string + replicas: number + +actor: + fields: + role: string + +snapshot: + fields: +""" + + +# ------------------------------------------------------------ +# 2. Policy +# ------------------------------------------------------------ + +policy_yaml = """ +version: 1 + +rules: + - id: block_large_scale + scope: + action: scale_service + when: + subject: + domain: action + field: replicas + operator: greater_than + value: + literal: 10 + effect: block +""" + + +# ------------------------------------------------------------ +# 3. Compile policy +# ------------------------------------------------------------ + +policy = Actra.from_strings(schema_yaml, policy_yaml) + + +# ------------------------------------------------------------ +# 4. Runtime +# ------------------------------------------------------------ + +runtime = ActraRuntime(policy) + +runtime.set_actor_resolver(lambda ctx: {"role": "operator"}) + + +# ------------------------------------------------------------ +# 5. Use allow() +# ------------------------------------------------------------ + +print("\n--- Allowed scaling ---") + +if runtime.allow("scale_service", service="search", replicas=5): + print("Scaling service to 5 replicas") + + +print("\n--- Blocked scaling ---") + +if not runtime.allow("scale_service", service="search", replicas=20): + print("Scaling request denied by policy") \ No newline at end of file diff --git a/examples/python/helpers/block_helper_example.py b/examples/python/helpers/block_helper_example.py new file mode 100644 index 0000000..eccefed --- /dev/null +++ b/examples/python/helpers/block_helper_example.py @@ -0,0 +1,92 @@ +""" +Actra Helper Example — block() + +The block() helper returns True if the policy decision +blocks the requested action + +This pattern is useful for guard checks before executing +sensitive operations +""" + +from actra import Actra +from actra.runtime import ActraRuntime + + +# ------------------------------------------------------------ +# 1. Schema +# ------------------------------------------------------------ + +schema_yaml = """ +version: 1 + +actions: + delete_cluster: + fields: + name: string + env: string + +actor: + fields: + role: string + +snapshot: + fields: +""" + + +# ------------------------------------------------------------ +# 2. Policy +# ------------------------------------------------------------ + +policy_yaml = """ +version: 1 + +rules: + - id: protect_prod_cluster + scope: + action: delete_cluster + when: + subject: + domain: action + field: env + operator: equals + value: + literal: "prod" + effect: block +""" + + +# ------------------------------------------------------------ +# 3. Compile policy +# ------------------------------------------------------------ + +policy = Actra.from_strings(schema_yaml, policy_yaml) + + +# ------------------------------------------------------------ +# 4. Runtime +# ------------------------------------------------------------ + +runtime = ActraRuntime(policy) + +runtime.set_actor_resolver(lambda ctx: {"role": "admin"}) + + +# ------------------------------------------------------------ +# 5. Use block() +# ------------------------------------------------------------ + +print("\n--- Attempting staging deletion ---") + +if runtime.block("delete_cluster", name="search-cluster", env="staging"): + print("Deletion blocked") +else: + print("Cluster deleted") + + +print("\n--- Attempting production deletion ---") + +if runtime.block("delete_cluster", name="prod-cluster", env="prod"): + print("Deletion blocked by policy") +else: + print("Cluster deleted") \ No newline at end of file diff --git a/examples/python/langchain/database_guardrail_agent.py b/examples/python/langchain/database_guardrail_agent.py new file mode 100644 index 0000000..74cc92e --- /dev/null +++ b/examples/python/langchain/database_guardrail_agent.py @@ -0,0 +1,153 @@ +""" +Actra Guardrail Example +======================= + +This example demonstrates how Actra prevents unsafe actions +triggered by an LLM. + +Scenario +-------- + +An AI system has access to infrastructure tools such as: + + - scaling services + - restarting systems + - deleting databases + +Deleting a production database is dangerous. Actra policies +ensure that unsafe actions are blocked deterministically. + +Flow +---- + +User > LLM > Tool Call > Actra Policy > Allow / Block + +If the policy blocks the action, the tool execution never occurs + +In a real agent system, the LLM would decide to call the tool +Here we simulate that decision by invoking the tool directly +""" + +from langchain.tools import tool +from langchain_community.llms.fake import FakeListLLM + +from actra import Actra, ActraRuntime, ActraPolicyError, ActraContext + + +# --------------------------------------------------------------------- +# 1. Schema +# --------------------------------------------------------------------- + +schema_yaml = """ +version: 1 + +actions: + delete_database: + fields: + database: string + environment: string + +actor: + fields: + role: string + +snapshot: + fields: + maintenance_window: boolean +""" + + +# --------------------------------------------------------------------- +# 2. Policy +# --------------------------------------------------------------------- + +policy_yaml = """ +version: 1 + +rules: + + - id: block_delete_production_database + scope: + action: delete_database + when: + subject: + domain: action + field: environment + operator: equals + value: + literal: "production" + effect: block +""" + + +policy = Actra.from_strings(schema_yaml, policy_yaml) +runtime = ActraRuntime(policy) + + +# --------------------------------------------------------------------- +# 3. Runtime Context +# --------------------------------------------------------------------- + +runtime.set_context_resolver( + lambda args, kwargs: ActraContext(user={"role": "engineer"}) +) + +runtime.set_actor_resolver( + lambda ctx: {"role": ctx.user.get("role")} +) + +runtime.set_snapshot_resolver( + lambda ctx: {"maintenance_window": False} +) + + +# --------------------------------------------------------------------- +# 4. Protected Tool +# --------------------------------------------------------------------- + +@tool +@runtime.admit() +def delete_database(database: str, environment: str) -> str: + """ + Delete a database. + + This tool is protected by Actra admission control. + """ + + return f"Database {database} deleted in {environment}" + + +# --------------------------------------------------------------------- +# 5. Fake LLM (simulated decision) +# --------------------------------------------------------------------- + +llm = FakeListLLM( + responses=[ + "Delete the users database in production" + ] +) + + +# --------------------------------------------------------------------- +# 6. Simulated LLM → Tool Execution +# --------------------------------------------------------------------- + +print("\n--- LLM Attempt ---\n") + +prompt = "Delete the users database in production" +response = llm.invoke(prompt) + +print("LLM request:", response) + +try: + result = delete_database.invoke( + { + "database": "users", + "environment": "production" + } + ) + + print("Result:", result) + +except ActraPolicyError: + print("Action blocked by Actra policy") \ No newline at end of file diff --git a/examples/python/langchain/refund_agent.py b/examples/python/langchain/refund_agent.py new file mode 100644 index 0000000..361de91 --- /dev/null +++ b/examples/python/langchain/refund_agent.py @@ -0,0 +1,110 @@ +""" +Actra + LangChain Example + +This example demonstrates how Actra can protect LangChain tools +executed by an LLM agent + +The LLM may decide to call a tool, but Actra evaluates a policy +decision before the tool executes + +In a real agent system, the LLM would decide to call the tool +Here we simulate that decision by invoking the tool directly +""" + +from langchain.tools import tool + +from actra import Actra, ActraRuntime, ActraContext + +# ------------------------------------------------------------ +# Schema +# ------------------------------------------------------------ + +schema_yaml = """ +version: 1 + +actions: + refund: + fields: + amount: number + +actor: + fields: + role: string + +snapshot: + fields: + fraud_flag: boolean +""" + +# ------------------------------------------------------------ +# Policy +# ------------------------------------------------------------ + +policy_yaml = """ +version: 1 + +rules: + - id: support_refund_limit + scope: + action: refund + when: + all: + - subject: + domain: action + field: amount + operator: greater_than + value: + literal: 1000 + - subject: + domain: actor + field: role + operator: equals + value: + literal: "support" + effect: block +""" + +policy = Actra.from_strings(schema_yaml, policy_yaml) +runtime = ActraRuntime(policy) + + +# ------------------------------------------------------------ +# Resolvers +# ------------------------------------------------------------ + +runtime.set_context_resolver( + lambda args, kwargs: ActraContext( + user={"role": "support"} + ) +) + +runtime.set_actor_resolver( + lambda ctx: {"role": ctx.user.get("role")} +) + +runtime.set_snapshot_resolver( + lambda ctx: {"fraud_flag": False} +) + + +# ------------------------------------------------------------ +# LangChain Tool +# ------------------------------------------------------------ + +@tool +@runtime.admit() +def refund(amount: int) -> str: + """Issue a refund to the customer.""" + return f"Refund executed: {amount}" + + +# ------------------------------------------------------------ +# Example Tool Calls +# ------------------------------------------------------------ + +print(refund.invoke({"amount": 200})) + +try: + print(refund.invoke({"amount": 2000})) +except Exception as e: + print("Blocked:", e) \ No newline at end of file diff --git a/examples/python/langchain/refund_llm_agent.py b/examples/python/langchain/refund_llm_agent.py new file mode 100644 index 0000000..1f117f8 --- /dev/null +++ b/examples/python/langchain/refund_llm_agent.py @@ -0,0 +1,161 @@ +""" +Actra + LangChain Guardrail Example +=================================== + +This example demonstrates how Actra can enforce deterministic +policy decisions for tools invoked by an LLM agent + +Flow: + +User > LLM > Tool Call > Actra Policy > Allow / Block + +If the policy blocks the action, the tool execution does not occur +and the LLM must respond accordingly + +This pattern is critical for safe agentic systems where AI may attempt +to perform actions such as: + + - issuing refunds + - modifying infrastructure + - deleting resources + - sending emails + +Actra acts as the deterministic guardrail controlling those actions. + +In a real agent system, the LLM would decide to call the tool +Here we simulate that decision by invoking the tool directly +""" + +from langchain.tools import tool +from langchain_community.llms.fake import FakeListLLM + +from actra import Actra, ActraRuntime, ActraPolicyError, ActraContext + +# ------------------------------------------------------------ +# Schema +# ------------------------------------------------------------ + +schema_yaml = """ +version: 1 + +actions: + refund: + fields: + amount: number + +actor: + fields: + role: string + +snapshot: + fields: + fraud_flag: boolean +""" + + +# ------------------------------------------------------------ +# Policy +# ------------------------------------------------------------ + +policy_yaml = """ +version: 1 + +rules: + - id: support_refund_limit + scope: + action: refund + when: + all: + - subject: + domain: action + field: amount + operator: greater_than + value: + literal: 1000 + - subject: + domain: actor + field: role + operator: equals + value: + literal: "support" + effect: block +""" + + +policy = Actra.from_strings(schema_yaml, policy_yaml) +runtime = ActraRuntime(policy) + +# ------------------------------------------------------------ +# Resolvers +# ------------------------------------------------------------ + +runtime.set_context_resolver( + lambda args, kwargs: ActraContext( + user={"role": "support"} + ) +) + +runtime.set_actor_resolver( + lambda ctx: {"role": ctx.user.get("role")} +) + +runtime.set_snapshot_resolver( + lambda ctx: {"fraud_flag": False} +) + + +# ------------------------------------------------------------ +# Protected Tool +# ------------------------------------------------------------ + +@tool +@runtime.admit() +def refund(amount: int) -> str: + """Refund tool protected by Actra.""" + return f"Refund executed: {amount}" + + +# ------------------------------------------------------------ +# Fake LLM +# ------------------------------------------------------------ + +llm = FakeListLLM( + responses=[ + "Call refund with amount=200", + "Call refund with amount=5000", + ] +) + + +# ------------------------------------------------------------ +# Simulated LLM Agent Behavior +# ------------------------------------------------------------ + +def run_agent(): + print("\n--- LLM Attempt 1 ---") + + response = llm.invoke("refund customer") + print(response) + + try: + print(refund.invoke({"amount": 200})) + except ActraPolicyError: + print("Blocked by policy") + + print("\n--- LLM Attempt 2 ---") + + response = llm.invoke("refund customer") + print(response) + + try: + print(refund.invoke({"amount": 5000})) + except ActraPolicyError: + print("Blocked by policy") + + +# ------------------------------------------------------------ +# Execute +# ------------------------------------------------------------ + +if __name__ == "__main__": + run_agent() \ No newline at end of file diff --git a/examples/python/mcp/fastmcp_basic.py b/examples/python/mcp/fastmcp_basic.py new file mode 100644 index 0000000..e758f56 --- /dev/null +++ b/examples/python/mcp/fastmcp_basic.py @@ -0,0 +1,211 @@ +""" +Actra + FastMCP Example +======================= + +This example demonstrates how to enforce Actra admission control +policies on MCP tools using FastMCP + +Actra evaluates a policy decision **before the tool executes**. +If the policy decision is `"block"`, the tool call is prevented +and the execution does not proceed + +Key concepts demonstrated in this example: + +- Defining an Actra schema +- Writing a simple policy rule +- Creating an Actra runtime +- Providing runtime context using resolvers +- Protecting MCP tools using the `@runtime.admit()` decorator + +This example intentionally keeps the runtime context minimal so it +can run in any MCP environment without requiring authentication +or external systems +""" + +from fastmcp import FastMCP + +from actra import Actra, ActraRuntime +from actra.integrations.mcp import ActraMCPContext + + +# --------------------------------------------------------------------- +# 1. MCP Server +# --------------------------------------------------------------------- + +# Create a FastMCP server instance. +# The name is used for identification in MCP tooling environments. +mcp = FastMCP("actra-example") + + +# --------------------------------------------------------------------- +# 2. Actra Schema +# --------------------------------------------------------------------- +# +# The schema defines the structure of the policy domains: +# +# action : parameters of the tool invocation +# actor : identity of the caller +# snapshot : external system state +# +# Policies are only allowed to reference fields defined here +# + +schema_yaml = """ +version: 1 + +actions: + refund: + fields: + amount: number + +actor: + fields: + role: string + +snapshot: + fields: + fraud_flag: boolean +""" + +# --------------------------------------------------------------------- +# 3. Policy Definition +# --------------------------------------------------------------------- +# +# Policy rule: +# +# Support agents are NOT allowed to issue refunds larger than 1000 +# + +policy_yaml = """ +version: 1 + +rules: + - id: support_refund_limit + scope: + action: refund + when: + all: + - subject: + domain: action + field: amount + operator: greater_than + value: + literal: 1000 + - subject: + domain: actor + field: role + operator: equals + value: + literal: "support" + effect: block +""" + +# --------------------------------------------------------------------- +# 4. Compile Policy and Create Runtime +# --------------------------------------------------------------------- + +policy = Actra.from_strings(schema_yaml, policy_yaml) +runtime = ActraRuntime(policy) + +# --------------------------------------------------------------------- +# 5. Runtime Context Resolvers +# --------------------------------------------------------------------- +# +# Resolvers provide dynamic runtime data used during policy evaluation. +# +# Actra separates policy logic from application frameworks +# Integrations (ActraMCPContext) are responsible for constructing the context object +# + +def resolve_context(args, kwargs) -> ActraMCPContext: + """ + Build the Actra context object for MCP tool invocations. + + In production environments this context may include: + + - authenticated user identity + - session metadata + - request identifiers + - tenant information + + For this example we provide a minimal context object + """ + return ActraMCPContext(user={}) + + +def resolve_actor(ctx: ActraMCPContext): + """ + Extract the actor domain from the MCP context + + The actor represents the entity invoking the tool + + In real systems this may come from: + + - authentication tokens + - MCP session identity + - API gateway headers + """ + return {"role": ctx.user.get("role", "support")} + + +def resolve_snapshot(ctx: ActraMCPContext): + """ + Provide external system state used by policies + + Snapshot data typically comes from: + + - databases + - feature flags + - fraud detection systems + - system configuration + + This example always returns `fraud_flag = False` + """ + return {"fraud_flag": False} + + +runtime.set_context_resolver(resolve_context) +runtime.set_actor_resolver(resolve_actor) +runtime.set_snapshot_resolver(resolve_snapshot) + +# --------------------------------------------------------------------- +# 6. MCP Tool Protected by Actra +# --------------------------------------------------------------------- +# +# The `@runtime.admit()` decorator enforces policy evaluation +# before the tool executes. +# +# If a policy decision results in `"block"`, the tool will not run +# + +@mcp.tool() +@runtime.admit() +def refund(amount: int): + """ + Issue a refund + + This tool is protected by Actra admission control + + Policy behavior: + - refunds <= 1000 : allowed + - refunds > 1000 by support agents : blocked + + Args: + amount: + Refund amount requested by the caller + + Returns: + A JSON-serializable response describing the result + """ + + return { + "message": f"Refund executed: {amount}" + } + + +# --------------------------------------------------------------------- +# 7. Run MCP Server +# --------------------------------------------------------------------- + +if __name__ == "__main__": + mcp.run() \ No newline at end of file diff --git a/examples/python/runtime/explain_call.py b/examples/python/runtime/explain_call.py new file mode 100644 index 0000000..8f457dd --- /dev/null +++ b/examples/python/runtime/explain_call.py @@ -0,0 +1,51 @@ +""" +Explain Call Example + +Demonstrates how to inspect policy evaluation for a +function call without executing the function. +""" + +from actra import Actra, ActraRuntime + + +schema_yaml = """ +version: 1 + +actions: + refund: + fields: + amount: number + +actor: + fields: + role: string + +snapshot: + fields: +""" + +policy_yaml = """ +version: 1 + +rules: + - id: block_large_refund + scope: + action: refund + when: + subject: + domain: action + field: amount + operator: greater_than + value: + literal: 1000 + effect: block +""" + +policy = Actra.from_strings(schema_yaml, policy_yaml) +runtime = ActraRuntime(policy) + +def refund(amount: int): + print("Refund executed:", amount) + +# This reproduces exactly what would happen if the function were executed with @runtime.admit +runtime.explain_call(refund, amount=1500) diff --git a/examples/python/runtime/explain_call_detailed.py b/examples/python/runtime/explain_call_detailed.py new file mode 100644 index 0000000..3dfd5f9 --- /dev/null +++ b/examples/python/runtime/explain_call_detailed.py @@ -0,0 +1,221 @@ +""" +Actra explain_call Example +-------------------------- + +This example demonstrates how to debug a policy decision using +`runtime.explain_call()` without executing the protected function + +`explain_call()` simulates the same evaluation flow used by the +`@runtime.admit` decorator, but instead of executing the function, +it prints a human-readable explanation of the policy decision + +Use this when you want to understand: + +• why a function call is allowed or blocked +• which rule triggered +• what runtime context was used during evaluation + +This is extremely useful when developing or debugging policies. +""" + +from actra import Actra, ActraRuntime + +# ------------------------------------------------------------ +# Schema +# ------------------------------------------------------------ + +# Defines the structure of data that policies can reference. +# +# Domains: +# action -> information about the operation being performed +# actor -> identity of the caller +# snapshot -> external system state +# + +schema_yaml = """ +version: 1 + +actions: + refund: + fields: + amount: number + +actor: + fields: + role: string + +snapshot: + fields: + fraud_flag: boolean +""" + +# ------------------------------------------------------------ +# Policy +# ------------------------------------------------------------ +# Rule: +# Support agents are not allowed to issue refunds greater +# than 1000. +# + +policy_yaml = """ +version: 1 + +rules: + - id: support_limit + scope: + action: refund + when: + all: + - subject: + domain: action + field: amount + operator: greater_than + value: + literal: 1000 + effect: block + + """ + +policy_yaml = """ +version: 1 + +rules: + - id: block_fraud_account + scope: + global: true + when: + subject: + domain: snapshot + field: fraud_flag + operator: equals + value: + literal: true + effect: block + + - id: block_large_refund_by_support + scope: + action: refund + when: + all: + - subject: + domain: action + field: amount + operator: greater_than + value: + literal: 1000 + - subject: + domain: actor + field: role + operator: equals + value: + literal: "support" + effect: block +""" + +# ------------------------------------------------------------ +# Compile policy +# ------------------------------------------------------------ + +policy = Actra.from_strings(schema_yaml, policy_yaml) + +# ------------------------------------------------------------ +# Create runtime +# ------------------------------------------------------------ +# The runtime connects application code to the policy engine. +# + +runtime = ActraRuntime(policy) + +# ------------------------------------------------------------ +# Example request context +# ------------------------------------------------------------ +# Many frameworks provide request context objects. +# The runtime resolvers can extract actor information from them. +# + +class RequestContext: + def __init__(self, role): + self.role = role + +# ------------------------------------------------------------ +# Actor resolver +# ------------------------------------------------------------ +# Converts the application context into an Actra actor object. +# + +runtime.set_actor_resolver( + lambda ctx: {"role": ctx.role} +) + +# ------------------------------------------------------------ +# Snapshot resolver +# ------------------------------------------------------------ +# Converts system state into Snapshot object. +# +runtime.set_snapshot_resolver( + lambda ctx: {"fraud_flag": False} +) + +# ------------------------------------------------------------ +# Protected function +# ------------------------------------------------------------ +# The function is protected using the Actra admission decorator. +# +# If the policy effect is "block", ActraPolicyError will be raised +# and the function will NOT execute. +# + +@runtime.admit(fields=["amount"]) +def refund(amount: int, ctx=None): + print("Refund executed:", amount) + +# ------------------------------------------------------------ +# Create a context where the user is a support agent +# ------------------------------------------------------------ +ctx = RequestContext(role="support") + +# ------------------------------------------------------------ +# Example 1: Allowed execution +# ------------------------------------------------------------ +# Refund below the limit should succeed. +# + +refund(500, ctx=ctx) + +# ------------------------------------------------------------ +# Example 2: Debug a blocked call +# ------------------------------------------------------------ +# Instead of executing the function (which would raise an error), +# we ask Actra to explain the policy decision. +# + +runtime.explain_call(refund, amount=2000, ctx=ctx) + +# ------------------------------------------------------------ +# Example 3: Debug a blocked call for Fraud Account +# ------------------------------------------------------------ +# Instead of executing the function (which would raise an error), +# we ask Actra to explain the policy decision. +# + +fraud_runtime = ActraRuntime(policy) + +# ------------------------------------------------------------ +# Actor resolver +# ------------------------------------------------------------ +# Converts the application context into an Actra actor object. +# +fraud_runtime.set_actor_resolver( + lambda ctx: {"role": ctx.role} +) + +# ------------------------------------------------------------ +# Snapshot resolver +# ------------------------------------------------------------ +# Converts system state into Snapshot object. +# +fraud_runtime.set_snapshot_resolver( + lambda ctx: {"fraud_flag": True} +) + +fraud_runtime.explain_call(refund, 50, ctx) \ No newline at end of file diff --git a/examples/python/runtime/extract_framework_context.py b/examples/python/runtime/extract_framework_context.py new file mode 100644 index 0000000..b84a523 --- /dev/null +++ b/examples/python/runtime/extract_framework_context.py @@ -0,0 +1,76 @@ +""" +Context Resolver Example + +Shows how runtime context can be extracted from +function arguments. +""" + +from actra import Actra, ActraRuntime + +schema_yaml = """ +version: 1 + +actions: + refund: + fields: + amount: number + +actor: + fields: + role: string + +snapshot: + fields: +""" + +policy_yaml = """ +version: 1 + +rules: + - id: support_limit + scope: + action: refund + when: + all: + - subject: + domain: action + field: amount + operator: greater_than + value: + literal: 1000 + - subject: + domain: actor + field: role + operator: equals + value: + literal: "support" + effect: block +""" + +policy = Actra.from_strings(schema_yaml, policy_yaml) +runtime = ActraRuntime(policy) + + +class RequestContext: + def __init__(self, role): + self.role = role + + +runtime.set_context_resolver( + lambda args, kwargs: kwargs.get("ctx") +) + +runtime.set_actor_resolver( + lambda ctx: {"role": ctx.role} +) + + +@runtime.admit(action_type="refund") +def refund(amount: int, ctx=None): + print("Refund executed:", amount) + + +ctx = RequestContext(role="support") + +refund(amount=200, ctx=ctx) +refund(amount=2000, ctx=ctx) diff --git a/examples/python/runtime/manual_evaluation_with_events.py b/examples/python/runtime/manual_evaluation_with_events.py index f60ecf8..88cef89 100644 --- a/examples/python/runtime/manual_evaluation_with_events.py +++ b/examples/python/runtime/manual_evaluation_with_events.py @@ -52,11 +52,11 @@ def observer(event:DecisionEvent): action = runtime.build_action( - func=lambda: None, action_type="deploy", args=(), kwargs={"env": "prod"}, - ctx=None + ctx=None, + func=None, ) runtime.evaluate(action) \ No newline at end of file diff --git a/scripts/run_all_examples.sh b/scripts/run_all_examples.sh new file mode 100755 index 0000000..218ea80 --- /dev/null +++ b/scripts/run_all_examples.sh @@ -0,0 +1,13 @@ +#!/bin/bash + +ROOT_DIR="$1" + +if [ -z "$ROOT_DIR" ]; then + echo "Usage: $0 " + exit 1 +fi + +find "$ROOT_DIR" -type f -name "*.py" | while read file; do + echo "Running $file" + python3 "$file" +done \ No newline at end of file diff --git a/sdk/python/README.md b/sdk/python/README.md index a2b3438..c710be5 100644 --- a/sdk/python/README.md +++ b/sdk/python/README.md @@ -1,7 +1,8 @@ # Actra Python SDK -Deterministic admission control and policy evaluation for state-changing -operations in automated and agentic systems. +**Action Admission Control for Automated Systems** + +Deterministic policy engine that decides whether automated actions are **allowed before they execute**. The **Actra Python SDK** provides a simple interface for loading policies and evaluating decisions using the Actra engine written in Rust. @@ -31,115 +32,37 @@ The package includes a compiled Rust engine, so no Rust toolchain is required du ## Quick Start ```python -import actra -from actra import ActraRuntime - -policy = actra.load_policy_from_file( - "schema.yaml", - "policy.yaml" -) - -runtime = ActraRuntime(policy) - -runtime.set_actor_resolver(lambda ctx: {"role": "admin"}) -runtime.set_snapshot_resolver(lambda ctx: {}) - -@runtime.admit() -def deploy(): - print("Deployment executed") - -deploy() - +@actra.admit() +def refund(amount): + ... ``` ---- - -## Loading Policies - -### From Files - -```python -import actra - -policy = actra.load_policy_from_file( - "schema.yaml", - "policy.yaml" -) +The rule lives in policy: + +```yaml +rules: + - id: block_large_refund + when: + subject: + domain: action + field: amount + operator: greater_than + value: + literal: 1000 + effect: block ``` -Optional governance configuration can also be provided: +```markdown +Result: -```python -policy = actra.load_policy_from_file( - "schema.yaml", - "policy.yaml", - "governance.yaml" -) +refund(200) > allowed +refund(1500) > blocked by policy ``` ---- - -### From Strings - -Useful for tests or dynamic environments. - -```python -policy = actra.load_policy_from_string( - schema_yaml, - policy_yaml -) -``` +Actra evaluates the policy **before the function executes** and blocks refunds greater than 1000. --- -## Evaluating Decisions - -Policies evaluate a request context. - -```python -decision = policy.evaluate({ - "action": {...}, - "actor": {...}, - "snapshot": {...} -}) -``` - -The context typically contains: - -| Field | Description | -| ---------- | ---------------------------- | -| `action` | operation being requested | -| `actor` | entity requesting the action | -| `snapshot` | current system state | - ---- - -## Policy Hash - -Every compiled policy has a deterministic hash. - -```python -policy.policy_hash() -``` - -This is useful for: - -* auditing -* verifying policy consistency - ---- - -## Engine Version - -Retrieve the underlying compiler version: - -```python -import actra - -actra.Actra.compiler_version() -``` - ---- ## Design Goals diff --git a/sdk/python/actra/__init__.py b/sdk/python/actra/__init__.py index facbecb..76afa6a 100644 --- a/sdk/python/actra/__init__.py +++ b/sdk/python/actra/__init__.py @@ -3,8 +3,9 @@ from .policy import Actra, Policy from .runtime import ActraRuntime from .types import Action, Actor, Snapshot, Decision, Context -from .errors import ActraError, ActraPolicyError +from .errors import ActraError, ActraPolicyError, ActraSchemaError from .events import DecisionEvent +from .integrations.mcp.context import ActraMCPContext as ActraContext try: __version__ = version("actra") @@ -19,9 +20,13 @@ "Actra", "Policy", "ActraRuntime", + "DecisionEvent", + "ActraContext", + + # Errors "ActraError", "ActraPolicyError", - "DecisionEvent", + "ActraSchemaError", # Runtime types "Action", diff --git a/sdk/python/actra/errors.py b/sdk/python/actra/errors.py index 035e513..fe41f96 100644 --- a/sdk/python/actra/errors.py +++ b/sdk/python/actra/errors.py @@ -131,4 +131,16 @@ def __repr__(self): f"ActraPolicyError(" f"action_type={self.action_type!r}, " f"decision={self.decision!r})" - ) \ No newline at end of file + ) + +class ActraSchemaError(ActraError): + """ + Raised when Actra schema YAML cannot be parsed. + + This typically indicates invalid YAML syntax or + an incorrectly formatted schema definition. + """ + + def __init__(self, message, original_error=None): + self.original_error = original_error + super().__init__(message) \ No newline at end of file diff --git a/sdk/python/actra/integrations/mcp/__init__.py b/sdk/python/actra/integrations/mcp/__init__.py new file mode 100644 index 0000000..95bc2a8 --- /dev/null +++ b/sdk/python/actra/integrations/mcp/__init__.py @@ -0,0 +1,7 @@ +from .context import ActraMCPContext +from .adapter import ActraMCPAdapter + +__all__ = [ + "ActraMCPContext", + "ActraMCPAdapter", +] \ No newline at end of file diff --git a/sdk/python/actra/integrations/mcp/adapter.py b/sdk/python/actra/integrations/mcp/adapter.py new file mode 100644 index 0000000..14e322f --- /dev/null +++ b/sdk/python/actra/integrations/mcp/adapter.py @@ -0,0 +1,108 @@ +""" +Actra MCP Adapter + +This module provides the adapter layer that connects MCP tool calls +to the Actra policy runtime + +The adapter translates MCP tool invocations into Actra actions and +delegates policy evaluation to `ActraRuntime` + +Responsibilities of the adapter include: + + - mapping MCP tool calls to Actra actions + - constructing action objects from tool arguments + - forwarding execution context to the runtime + - evaluating policy decisions before tool execution + +The adapter does not implement policy logic itself. Instead it relies +on the core runtime for action construction and evaluation + +Execution flow: + + 1. MCP Tool Call + 2. ActraMCP Adapter + 3. ActraRuntime.build_action() + 4. ActraRuntime.evaluate() + 5. Policy Decision + +If a policy decision results in `"block"`, the adapter raises `ActraPolicyError` +and prevents tool execution + +Example usage: + + runtime = ActraRuntime(policy) + mcp = ActraMCPAdapter(runtime) + decision = mcp.evaluate_tool( + tool_name="refund", + arguments={"amount": 200}, + ctx=context + ) + +The adapter is intentionally lightweight so that ActraRuntime remains +the single orchestration layer for policy evaluation across all +integrations +""" + +from typing import Dict, Any, Optional + +from actra.errors import ActraPolicyError +from actra.types import Context + +class ActraMCPAdapter: + """ + Adapter connecting MCP tool calls to ActraRuntime + + The adapter converts MCP tool invocations into Actra actions and + delegates policy evaluation to the runtime + + It does not implement policy logic itself and remains a thin layer + translating MCP inputs into Actra runtime calls + """ + + def __init__(self, runtime): + self.runtime = runtime + + def evaluate_tool( + self, + tool_name: str, + arguments: Dict[str,Any], + ctx: Optional[Context] = None): + """ + Evaluate whether an MCP tool call should be allowed + + Args: + tool_name: + Name of the MCP tool being invoked + + arguments: + Dictionary containing tool arguments + + ctx: + MCP execution context + + Returns: + Policy decision dictionary + + Raises: + ActraPolicyError: + If the policy blocks the action. + """ + action = self.runtime.build_action( + action_type=tool_name, + args=(), + kwargs=arguments, + ctx=ctx + ) + + decision = self.runtime.evaluate(action, ctx) + if decision.get("effect") == "block": + context = self.runtime.build_context(action, ctx) + + raise ActraPolicyError( + action_type=tool_name, + decision=decision, + context=context + ) + return decision + +__all__ = ["ActraMCPAdapter"] diff --git a/sdk/python/actra/integrations/mcp/context.py b/sdk/python/actra/integrations/mcp/context.py new file mode 100644 index 0000000..66fbfe9 --- /dev/null +++ b/sdk/python/actra/integrations/mcp/context.py @@ -0,0 +1,80 @@ +""" +MCP Context Model for Actra Integrations + +This module defines the context object used when evaluating Actra +policies for MCP tool calls + +Actra separates policy evaluation from application frameworks +Integrations are responsible for constructing a context object that +contains any runtime information required by policy resolvers + +In MCP environments this context typically includes: + + - user identity + - session metadata + - request information + - tool execution environment + +The context object is passed to `ActraRuntime.evaluate()` during policy evaluation +and is used by configured resolvers to populate the policy domains: + + actor + snapshot + +Example: + + ctx = MCPContext( + user={"id": "123", "role": "support"}, + metadata={"session_id": "abc"} + ) + +Resolvers can then extract policy inputs from the context: + + runtime.set_actor_resolver(lambda ctx: {"role": ctx.user["role"]}) + +The Actra runtime itself remains framework-agnostic and does not +interpret the context object directly. Instead, integrations define +how the context should be structured and how resolvers derive the +required policy inputs + +This design keeps the core Actra runtime independent from MCP and other integrations while +allowing flexible policy evaluation across different environments +""" + +from dataclasses import dataclass +from typing import Optional, Dict, Any + +@dataclass +class ActraMCPContext: + """ + Context object passed to ActraRuntime when evaluating MCP tool calls + + The context carries runtime information about the MCP request + environment such as user identity, metadata, and request details + + Integrations populate this object and pass it to the Actra runtime, + where configured resolvers derive the policy domains (`actor`, + `snapshot`) from the context + + Attributes: + user: + Information about the user invoking the MCP tool + + metadata: + Arbitrary session or execution metadata + + request: + Raw MCP request data or tool invocation details + + Example: + ctx = MCPContext( + user={"id": "123", "role": "support"}, + metadata={"session": "abc"} + ) + """ + + user: Optional[Dict[str, Any]] = None + metadata: Optional[Dict[str, Any]] = None + request: Optional[Dict[str, Any]] = None + +__all__ = ["ActraMCPContext"] \ No newline at end of file diff --git a/sdk/python/actra/integrations/mcp/server.py b/sdk/python/actra/integrations/mcp/server.py new file mode 100644 index 0000000..ec1cb09 --- /dev/null +++ b/sdk/python/actra/integrations/mcp/server.py @@ -0,0 +1,105 @@ +""" +MCP Server Utilities for Actra + +This module provides helper utilities for building MCP-compatible +tool servers that integrate with Actra admission control + +The server layer is responsible for: + + - registering tool functions + - receiving MCP tool invocation requests + - constructing execution context objects + - delegating policy evaluation to the Actra MCP adapter + - executing tools if policies allow the operation + +The server does not contain any policy logic. Instead it coordinates +between the MCP environment and the Actra runtime + +The MCP Server handles : + 1. Tool registration : expose Python functions as MCP tools + 2. Request handling : receive MCP tool calls + 3. Argument parsing : convert tool arguments to Python kwargs + 4. Context creation : create `ctx` object from MCP request + 5. Tool execution : call the Python function + +Typical execution flow: + 1. MCP Request + 2. Tool Lookup + 3. Context Construction + 4. Actra Policy Evaluation + 5. Allow, Execute Tool + 6. Block, Raise Error + +This separation keeps Actra's policy engine independent from the MCP +server implementation while enabling admission control for tool +execution +""" + +from typing import Callable, Dict, Any, Optional + +from actra.integrations.mcp.adapter import ActraMCPAdapter +from actra.types import Context + +class ActraMCPServer: + """ + Minimal MCP tool server for Actra integration. + + This server registers Python functions as MCP tools and ensures that + Actra policies are evaluated before executing those tools + """ + + def __init__(self, adapter: ActraMCPAdapter): + self.adapter = adapter + self.tools: Dict[str, Callable] = {} + + def tool(self, name: Optional[str] = None): + """ + Decorator used to register a tool. + + Example: + + @server.tool() + def refund(amount: int): + ... + """ + def decorator(func: Callable): + tool_name = name or func.__name__ + self.tools[tool_name] = func + return func + return decorator + + def handle_request( + self, + tool_name: str, + arguments: Dict[str, Any], + ctx: Optional[Context] = None): + """ + Execute an MCP tool call after policy evaluation + + Args: + tool_name: + Name of the MCP tool being invoked + + arguments: + Tool input arguments + + ctx: + Optional execution context passed to the Actra runtime + + Returns: + Result returned by the tool function + + Raises: + ActraPolicyError: + If the policy blocks the operation + """ + tool = self.tools.get(tool_name) + if not tool: + raise ValueError(f"Unknown tool: {tool_name}") + + # Adapter will raise ActraPolicyError if blocked + self.adapter.evaluate_tool(tool_name, arguments, ctx) + + return tool(**arguments) + +__all__ = ["ActraMCPServer"] \ No newline at end of file diff --git a/sdk/python/actra/policy.py b/sdk/python/actra/policy.py index 5a79da4..99c6657 100644 --- a/sdk/python/actra/policy.py +++ b/sdk/python/actra/policy.py @@ -1,8 +1,11 @@ from pathlib import Path from typing import Optional +import yaml + # Rust binding from .actra import PyActra as _RustActra from .types import Action, Actor, Snapshot, Decision, ActionInput, PathType +from .errors import ActraSchemaError class Policy: """ @@ -29,7 +32,7 @@ class Policy: Policies are deterministic and side-effect free. """ - def __init__(self, engine: _RustActra): + def __init__(self, engine: _RustActra, schema_yaml: Optional[str] = None): """ Internal constructor. @@ -41,6 +44,10 @@ def __init__(self, engine: _RustActra): Actra.from_directory(...) """ self._engine = engine + try: + self._schema = yaml.safe_load(schema_yaml) if schema_yaml else None + except yaml.YAMLError as e: + raise ActraSchemaError("Invalid Actra schema YAML") from e # ------------------------------------------------------------------ # Runtime API @@ -278,7 +285,7 @@ def from_strings( A compiled `Policy` object ready for evaluation. """ engine = _RustActra(schema_yaml, policy_yaml, governance_yaml) - return Policy(engine) + return Policy(engine, schema_yaml) @staticmethod def from_files( @@ -321,7 +328,7 @@ def from_files( governance_yaml = governance_file.read_text(encoding="utf-8") engine = _RustActra(schema_yaml, policy_yaml, governance_yaml) - return Policy(engine) + return Policy(engine, schema_yaml) @staticmethod def compiler_version() -> str: diff --git a/sdk/python/actra/runtime.py b/sdk/python/actra/runtime.py index a0cf10d..36ff9ed 100644 --- a/sdk/python/actra/runtime.py +++ b/sdk/python/actra/runtime.py @@ -91,7 +91,7 @@ def set_decision_observer(self, fn: Callable[[DecisionEvent], None]) -> None: Example: def observer(event): - print(event.effect, event.rule_id) + print(event.effect, event.matched_rule) runtime.set_decision_observer(observer) """ @@ -298,8 +298,57 @@ def _emit_decision_event(self, decision, action, context, duration_ms): ) self._decision_observer(event) + + def _bind_arguments(self, func: Callable, args: Tuple, kwargs: Dict[str, Any]) -> Dict[str, Any]: + """ + Bind positional and keyword arguments to function parameters. + + Returns a dictionary mapping parameter names to values. + """ + if func is None: + return dict(kwargs) + + sig = inspect.signature(func) + bound = sig.bind_partial(*args, **kwargs) + bound.apply_defaults() + return dict(bound.arguments) + + def allow(self, action_type: str, ctx: Context = None, **fields) -> bool: + """ + Convenience helper that returns True if the policy allows the action. + """ + action = self.action(action_type, **fields) + decision = self.evaluate(action, ctx) + + return decision.get("effect") == "allow" + + def block(self, action_type: str, ctx: Context = None, **fields) -> bool: + """ + Convenience helper that returns True if the policy blocks the action. + """ + + action = self.action(action_type, **fields) + decision = self.evaluate(action, ctx) + + return decision.get("effect") == "block" # Action construction + def action(self, action_type: str, **fields) -> Action: + """ + Construct a policy action using runtime schema validation. + + This helper is intended for direct programmatic policy checks + without needing a Python function signature. + + Example: + + runtime.action("deploy", env="prod") + """ + + return { + "type": action_type, + **fields + } def build_action(self, action_type: str, @@ -323,7 +372,27 @@ def build_action(self, This prevents internal parameters (such as request context, framework metadata, etc.) from leaking into the policy engine + The following is resolution priority : + 1. action_builder override + 2. runtime action_resolver override + 3. explicit fields parameter + 4. function signature filtering + 4.1 Refer schema + 5. fallback kwargs + Args: + action_type: + Logical action name used for policy evaluation + + args: + Positional arguments supplied to the function + + kwargs: + Keyword arguments supplied to the function + + ctx: + Optional execution context used by resolvers + func: Optional Function whose signature defines the allowed action fields. The function is **not executed**. It is only inspected to @@ -337,18 +406,6 @@ def build_action(self, (for example APIs, MCP tools, message queues) may pass `None`. - action_type: - Logical action name used for policy evaluation - - args: - Positional arguments supplied to the function - - kwargs: - Keyword arguments supplied to the function - - ctx: - Optional execution context used by resolvers - fields: Optional list restricting which keyword arguments are included in the action object @@ -376,14 +433,36 @@ def build_action(self, # Default: include kwargs but ignore internal parameters if func: sig = inspect.signature(func) - allowed_fields = set(sig.parameters) + func_fields = set(sig.parameters) + + schema_fields = None + schema = self.policy._schema + + if schema: + actions = schema.get("actions", {}) + action_schema = actions.get(action_type) + if action_schema: + schema_fields = set(action_schema.get("fields", {}).keys()) + + if schema_fields: + allowed_fields = func_fields & schema_fields + else: + allowed_fields = func_fields + + if args: + bound = sig.bind_partial(*args, **kwargs) + bound.apply_defaults() + bound_args = dict(bound.arguments) + else: + bound_args = dict(kwargs) + action_fields = { k: v - for k, v in kwargs.items() + for k, v in bound_args.items() if k in allowed_fields } else: - # No function available — trust provided kwargs + # No function available — trust provided kwargs API/MCP/Queue etc action_fields = dict(kwargs) return { "type": action_type, @@ -442,12 +521,18 @@ def _enforce_policy( This method is used internally by the `admit` decorator """ - ctx = self.resolve_context(args, kwargs) - act = self.resolve_action_type(func, args, kwargs, action_type) + bound_args = self._bind_arguments(func, args, kwargs) + ctx = self.resolve_context(args, bound_args) + + # RULE: + # Once _bind_arguments() is called, + # positional args must not be used again. + # + act = self.resolve_action_type(func, args, bound_args, action_type) action = self.build_action( act, - args, - kwargs, + (), #Positional args no more needed + bound_args, ctx, func=func, fields=fields, @@ -492,20 +577,31 @@ def explain(self, action: Action, ctx: Context = None) -> Decision: context = self.build_context(action, ctx) return self.policy.explain(context) - def explain_call(self, func: Callable, *args, **kwargs) -> Decision: + def explain_call( + self, + func: Callable, + *args, + action_type: Optional[str] = None, + ctx: Optional[Any] = None, + **kwargs + ) -> Decision: """ Explain the policy decision for a function call without executing it. This helper reconstructs the policy evaluation flow used by the `@runtime.admit` decorator and prints a human-readable explanation - of the decision. + of the resulting decision. + + Unlike the decorator, this method **does not execute the function**. + It only simulates the policy evaluation using the provided inputs. - It is particularly useful for: + This is particularly useful for: - debugging policy behavior - understanding why a rule triggered - testing policy inputs interactively - verifying runtime resolvers + - exploring policy decisions without modifying application code The method performs the following steps: @@ -515,14 +611,46 @@ def explain_call(self, func: Callable, *args, **kwargs) -> Decision: 4. Construct the full evaluation context 5. Invoke `Policy.explain()` to display the decision - Unlike the decorator, this method does not execute the function. + Action Type Resolution + ---------------------- + + The action type used for evaluation is determined in the following order: + + 1. Explicit `action_type` provided to `explain_call` + 2. A configured `action_type_resolver` + 3. The function name + + This allows `explain_call` to work even when the function name does not + match the action name defined in the schema Args: func: - The function protected by Actra admission control. + The function associated with the action being evaluated. + The function is **not executed**. It is only inspected to + determine the action structure *args: - Positional arguments that would be passed to the function. + Positional arguments that would be passed to the function + + action_type: + Optional explicit action name used for policy evaluation + This is useful when the function name differs from the + action defined in the schema + + ctx: + Optional execution context object used by runtime resolvers + + The context is passed to the configured actor and snapshot + resolvers to construct the evaluation context + + Example: + + runtime.set_actor_resolver( + lambda ctx: {"role": ctx.role} + ) + + If not provided explicitly, Actra will attempt to resolve + the context from the function arguments. **kwargs: Keyword arguments that would be passed to the function. @@ -534,6 +662,17 @@ def explain_call(self, func: Callable, *args, **kwargs) -> Decision: runtime.explain_call(refund, amount=1500) + Example with explicit action mapping: + + runtime.explain_call( + scale, + action_type="scale_service", + service="search-api", + replicas=20, + environment="staging", + ctx=operator + ) + Example output: Actra Decision @@ -551,18 +690,27 @@ def explain_call(self, func: Callable, *args, **kwargs) -> Decision: Result: effect: block - rule_id: block_large_refund + matched_rule: block_large_refund """ - ctx = self.resolve_context(args, kwargs) - act = self.resolve_action_type(func, args, kwargs, None) + bound_args = self._bind_arguments(func, args, kwargs) + + if ctx is None: + ctx = self.resolve_context(args, bound_args) + + # RULE: + # Once _bind_arguments() is called, + # positional args must not be used again. + # + + act = self.resolve_action_type(func, args, bound_args, action_type) action = self.build_action( - func, act, - args, - kwargs, - ctx + (), #Positional args no more needed + bound_args, + ctx, + func=func ) return self.explain(action, ctx) diff --git a/sdk/python/pyproject.toml b/sdk/python/pyproject.toml index 6369ec6..05754eb 100644 --- a/sdk/python/pyproject.toml +++ b/sdk/python/pyproject.toml @@ -31,7 +31,9 @@ classifiers = [ "Topic :: Security" ] -dependencies = [] +dependencies = [ + "PyYAML>=6.0.2" +] [project.urls] Homepage = "https://github.com/getactra/actra"