Skip to content
Anlyon
Esc
↑↓navigate↵open⌘Jpreview
On this page

CrewAI: hosted tools and human approval

Call an Anlyon action from a CrewAI tool, wait for an approver outside the crew, keep provider credentials out of the crew's process, and report the receipt grade.

CrewAI runs the crew. Anlyon executes the named external action. As of 2026-10-18, a CrewAI PRE_TOOL_CALL hook runs in the crew’s process: it can ask for a yes with ctx.request_human_input() and block the call by raising HookAborted, and it does not hold credentials. A hook’s yes does not approve a pending Anlyon invocation. See CrewAI’s tool call hooks.

Prerequisites

Complete the quickstart in a test environment. Define a governed action, for example refund-payment on the stripe.refund adapter with requires_approval=True, using an operator credential outside the crew. See Governed effects for the adapter config and Impact limits to cap it.

The crew needs an Anlyon agent credential with actions:invoke, actions:read and approvals:read. It does not need the provider key or approvals:decide.

pip install crewai anlyon

A tool that invokes the action

A CrewAI tool is a BaseTool with an args_schema and a _run method. _run invokes the action, waits for the decision, reads the same invocation until it finishes, and returns one plain sentence for the model.

import os
import time
from typing import Type

from anlyon import Client
from crewai.tools import BaseTool
from pydantic import BaseModel, Field

anlyon = Client(api_key=os.environ["ANLYON_AGENT_KEY"])
anlyon.auth.require_scopes(
    ["actions:invoke", "actions:read", "approvals:read"],
    forbidden=["actions:write", "secrets:write", "approvals:decide"],
)


class RefundInput(BaseModel):
    order_id: str = Field(..., description="Your order id. One refund per order.")
    payment_intent: str = Field(..., description="Stripe payment intent id, starting with pi_.")
    amount: int = Field(..., description="Amount to refund, in cents.")


class RefundPaymentTool(BaseTool):
    name: str = "refund_payment"
    description: str = "Refund a Stripe payment in USD. Amount is in cents. Every refund waits for an approver."
    args_schema: Type[BaseModel] = RefundInput

    def _run(self, order_id: str, payment_intent: str, amount: int) -> str:
        result = anlyon.actions.invoke(
            "refund-payment",
            {"paymentIntent": payment_intent, "amount": amount, "currency": "usd"},
            operation_key=f"refund:{order_id}",
        )
        invocation = result.data
        if invocation["pendingApproval"]:
            decided = anlyon.approvals.wait_for_decision(invocation["approvalId"], timeout_seconds=600)
            if decided["status"] == "pending":
                return "Waiting for an approver. Do not retry."
            for _ in range(30):
                invocation = anlyon.actions.invocation(invocation["id"]).data
                if invocation["status"] not in ("pending_approval", "running"):
                    break
                time.sleep(2)
        return f"Refund {invocation['id']}: grade {invocation.get('grade') or 'pending'}. Do not retry."

Pass the tool to an Agent(tools=[RefundPaymentTool()]) as usual. Derive operation_key from your own business record, not from the model. The same key and request return the existing effect instead of a second refund.

What the tool should tell the model

Grade Tool behavior
confirmed Report success. A read-back matched the request
acknowledged Report that the provider accepted it. Do not retry
unknown No usable response. Anlyon does not send it again. Do not retry; an operator settles it
failed Report the provider’s rejection
refused Stop. Anlyon refused it at dispatch, for example on an exhausted limit or an expired approval
denied Stop. Do not resubmit to bypass the decision
pending Report that it is not finished. Read the same invocation later

Hooks and Anlyon approvals together

A PRE_TOOL_CALL hook is useful as a check with whoever is running the crew. The Anlyon approval is the decision on the production call: bound to the exact request, decided by a credential other than the one that asked. Keep provider keys out of the crew’s environment, or a tool can call the provider directly and skip both. Task(human_input=True) reviews the task’s final answer, not a tool call.

A CrewAI tool that calls a provider directly is not routed through Anlyon and is not governed by it. For a call you cannot move yet, approvals.gate() waits for a decision and then runs your own function with your own credential.

See the full CrewAI example, receipts and grades, and the Python SDK.

Was this page helpful?