diff --git a/examples/agent-frameworks/crewai/README.md b/examples/agent-frameworks/crewai/README.md new file mode 100644 index 0000000..28b1f34 --- /dev/null +++ b/examples/agent-frameworks/crewai/README.md @@ -0,0 +1,63 @@ +# APort CrewAI Task Verification Decorator + +This example adds an `@aport_verify` decorator that verifies a CrewAI task with APort before the task executes. + +The decorator protects synchronous functions, bound methods, and asynchronous task callables, which covers common CrewAI task patterns. + +## Install + +```bash +python -m venv .venv +. .venv/bin/activate +pip install -e ".[test]" +``` + +Optional CrewAI runtime: + +```bash +pip install -e ".[crewai]" +``` + +Optional APort SDK package: + +```bash +pip install -e ".[sdk]" +``` + +## Configure + +```bash +export APORT_API_KEY=your_api_key_here +export APORT_API_URL=https://aport.io/api/verify +``` + +## Usage + +```python +from aport_crewai import aport_verify + + +@aport_verify( + "finance.payment.refund.v1", + task_type="refund_approval", + context=lambda order_id, amount: {"order_id": order_id, "amount": amount}, +) +def approve_refund(order_id: str, amount: float) -> str: + return f"Refund approved for {order_id}: ${amount:.2f}" +``` + +If APort denies the request, the decorator raises `APortVerificationError` before the task body runs. + +## Example Task Types + +- `refund_approval` for finance workflows. +- `data_export` for customer data exports. +- `repo_merge` for repository operations. +- `generic` for custom CrewAI tasks. + +## Test + +```bash +pip install -e ".[test]" +pytest +``` diff --git a/examples/agent-frameworks/crewai/examples/crew.py b/examples/agent-frameworks/crewai/examples/crew.py new file mode 100644 index 0000000..fb4b811 --- /dev/null +++ b/examples/agent-frameworks/crewai/examples/crew.py @@ -0,0 +1,28 @@ +from aport_crewai import aport_verify + + +@aport_verify( + "finance.payment.refund.v1", + task_type="refund_approval", + context=lambda order_id, amount: {"order_id": order_id, "amount": amount}, +) +def approve_refund(order_id: str, amount: float) -> str: + return f"Refund approved for {order_id}: ${amount:.2f}" + + +@aport_verify( + "data.export.v1", + task_type="data_export", + context={"dataset": "customers", "contains_pii": True}, +) +def export_customer_rows(limit: int) -> str: + return f"Exporting {limit} customer rows" + + +@aport_verify("code.repository.merge.v1", task_type="repo_merge") +async def merge_pull_request(repo: str, pull_request: int) -> str: + return f"Merging {repo}#{pull_request}" + + +if __name__ == "__main__": + print(approve_refund("order_123", 25.0)) diff --git a/examples/agent-frameworks/crewai/pyproject.toml b/examples/agent-frameworks/crewai/pyproject.toml new file mode 100644 index 0000000..bdd98f9 --- /dev/null +++ b/examples/agent-frameworks/crewai/pyproject.toml @@ -0,0 +1,31 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "aport-crewai-decorator-example" +version = "0.1.0" +description = "CrewAI task verification decorator example for APort" +requires-python = ">=3.10" +dependencies = [ + "requests>=2.32.0", +] + +[project.optional-dependencies] +crewai = [ + "crewai>=0.80.0", +] +sdk = [ + "aporthq-sdk-python>=0.1.0", +] +test = [ + "pytest>=8.0.0", + "pytest-asyncio>=0.23.0", +] + +[tool.pytest.ini_options] +pythonpath = ["src"] +testpaths = ["tests"] + +[tool.setuptools.packages.find] +where = ["src"] diff --git a/examples/agent-frameworks/crewai/src/aport_crewai/__init__.py b/examples/agent-frameworks/crewai/src/aport_crewai/__init__.py new file mode 100644 index 0000000..11b26f0 --- /dev/null +++ b/examples/agent-frameworks/crewai/src/aport_crewai/__init__.py @@ -0,0 +1,3 @@ +from .decorator import APortDecision, APortVerificationError, aport_verify + +__all__ = ["APortDecision", "APortVerificationError", "aport_verify"] diff --git a/examples/agent-frameworks/crewai/src/aport_crewai/decorator.py b/examples/agent-frameworks/crewai/src/aport_crewai/decorator.py new file mode 100644 index 0000000..bb5e4ec --- /dev/null +++ b/examples/agent-frameworks/crewai/src/aport_crewai/decorator.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +from dataclasses import dataclass +from functools import wraps +import inspect +import os +from typing import Any, Callable, Mapping + +import requests + + +class APortVerificationError(RuntimeError): + """Raised when APort denies or cannot verify a CrewAI task.""" + + +@dataclass(frozen=True) +class APortDecision: + allow: bool + reasons: tuple[str, ...] = () + decision_id: str | None = None + + @classmethod + def from_response(cls, response: Mapping[str, Any]) -> "APortDecision": + decision = response.get("data", {}).get("decision") or response.get("decision") or response + reasons = tuple( + reason.get("message") or reason.get("code") or str(reason) + for reason in decision.get("reasons", []) + ) + return cls( + allow=bool(decision.get("allow")), + reasons=reasons, + decision_id=decision.get("decision_id"), + ) + + +class APortClient: + def __init__( + self, + api_key: str | None = None, + api_url: str | None = None, + timeout: int = 10, + ) -> None: + self.api_key = api_key if api_key is not None else os.getenv("APORT_API_KEY", "") + self.api_url = api_url or os.getenv("APORT_API_URL", "https://aport.io/api/verify") + self.timeout = timeout + + def verify(self, policy_pack: str, context: Mapping[str, Any]) -> APortDecision: + response = requests.post( + self.api_url, + headers={ + "authorization": f"Bearer {self.api_key}", + "content-type": "application/json", + }, + json={"policy_pack": policy_pack, "context": dict(context)}, + timeout=self.timeout, + ) + response.raise_for_status() + return APortDecision.from_response(response.json()) + + +def _serializable(value: Any) -> Any: + if value is None or isinstance(value, (str, int, float, bool)): + return value + if isinstance(value, Mapping): + return {str(key): _serializable(item) for key, item in value.items()} + if isinstance(value, (list, tuple, set)): + return [_serializable(item) for item in value] + return repr(value) + + +def build_task_context( + task_type: str, + func: Callable[..., Any], + args: tuple[Any, ...], + kwargs: Mapping[str, Any], + extra_context: Mapping[str, Any] | Callable[..., Mapping[str, Any]] | None, +) -> dict[str, Any]: + context: dict[str, Any] = { + "task_type": task_type, + "callable": f"{func.__module__}.{func.__qualname__}", + "args": _serializable(args), + "kwargs": _serializable(kwargs), + } + + if extra_context: + additional = extra_context(*args, **kwargs) if callable(extra_context) else extra_context + context.update(_serializable(additional)) + + return context + + +def _raise_denied(decision: APortDecision) -> None: + reason_text = "; ".join(decision.reasons) if decision.reasons else "APort denied task execution" + if decision.decision_id: + reason_text = f"{reason_text} (decision {decision.decision_id})" + raise APortVerificationError(reason_text) + + +def aport_verify( + policy_pack: str, + *, + task_type: str = "generic", + client: Any | None = None, + context: Mapping[str, Any] | Callable[..., Mapping[str, Any]] | None = None, +) -> Callable[[Callable[..., Any]], Callable[..., Any]]: + """Verify a CrewAI task before execution. + + The decorator works with normal functions, bound methods, and async task + callables, so it can protect CrewAI tasks implemented with different task + patterns. + """ + + verifier = client or APortClient() + + def decorator(func: Callable[..., Any]) -> Callable[..., Any]: + if inspect.iscoroutinefunction(func): + + @wraps(func) + async def async_wrapper(*args: Any, **kwargs: Any) -> Any: + task_context = build_task_context(task_type, func, args, kwargs, context) + decision = verifier.verify(policy_pack, task_context) + if not decision.allow: + _raise_denied(decision) + return await func(*args, **kwargs) + + return async_wrapper + + @wraps(func) + def wrapper(*args: Any, **kwargs: Any) -> Any: + task_context = build_task_context(task_type, func, args, kwargs, context) + decision = verifier.verify(policy_pack, task_context) + if not decision.allow: + _raise_denied(decision) + return func(*args, **kwargs) + + return wrapper + + return decorator diff --git a/examples/agent-frameworks/crewai/tests/test_decorator.py b/examples/agent-frameworks/crewai/tests/test_decorator.py new file mode 100644 index 0000000..a7b8f70 --- /dev/null +++ b/examples/agent-frameworks/crewai/tests/test_decorator.py @@ -0,0 +1,80 @@ +import pytest + +from aport_crewai import APortDecision, APortVerificationError, aport_verify +from aport_crewai.decorator import build_task_context + + +class FakeClient: + def __init__(self, decision): + self.decision = decision + self.calls = [] + + def verify(self, policy_pack, context): + self.calls.append((policy_pack, context)) + return self.decision + + +def test_allows_sync_task_after_verification(): + client = FakeClient(APortDecision(allow=True, decision_id="dec_allow")) + + @aport_verify("finance.payment.refund.v1", task_type="refund", client=client) + def refund(amount): + return f"refunded {amount}" + + assert refund(25) == "refunded 25" + assert client.calls[0][0] == "finance.payment.refund.v1" + assert client.calls[0][1]["task_type"] == "refund" + assert client.calls[0][1]["args"] == [25] + + +def test_denies_sync_task_before_execution(): + client = FakeClient(APortDecision(allow=False, reasons=("limit exceeded",), decision_id="dec_deny")) + executed = False + + @aport_verify("finance.payment.refund.v1", client=client) + def refund(): + nonlocal executed + executed = True + + with pytest.raises(APortVerificationError, match="limit exceeded"): + refund() + + assert executed is False + + +@pytest.mark.asyncio +async def test_allows_async_task_after_verification(): + client = FakeClient(APortDecision(allow=True)) + + @aport_verify("code.repository.merge.v1", task_type="repo_merge", client=client) + async def merge(repo, number): + return f"{repo}#{number}" + + assert await merge("aporthq/aport-integrations", 117) == "aporthq/aport-integrations#117" + assert client.calls[0][1]["task_type"] == "repo_merge" + + +def test_includes_static_and_dynamic_context(): + client = FakeClient(APortDecision(allow=True)) + + @aport_verify( + "data.export.v1", + client=client, + context=lambda limit: {"dataset": "customers", "limit": limit}, + ) + def export(limit): + return limit + + export(100) + assert client.calls[0][1]["dataset"] == "customers" + assert client.calls[0][1]["limit"] == 100 + + +def test_build_task_context_serializes_non_json_values(): + def task(value): + return value + + context = build_task_context("generic", task, ({object()},), {}, {"extra": object()}) + assert context["task_type"] == "generic" + assert isinstance(context["args"][0][0], str) + assert isinstance(context["extra"], str)