Plan Approval Callback

Async plan-approval callbacks that work from inside a running event loop

ApprovalCallback gates a plan before it runs, and its async approve_fn now works even when called from inside a running event loop — such as a FastAPI async route.

Quick Start

1. Attach an async approval function

   Pass an async approve_fn to ApprovalCallback. It returns True to approve, False to reject.

   from praisonaiagents.planning import ApprovalCallback
   
   async def review_plan(plan) -> bool:
       # e.g. call a Slack / email / DB check
       return await my_async_approval_check(plan)
   
   approval = ApprovalCallback(approve_fn=review_plan)

2. Works inside a running loop

   The synchronous __call__ can now be reached from inside a running loop without crashing — the async approve_fn runs on the caller’s loop.

   import asyncio
   from praisonaiagents.planning import ApprovalCallback, Plan
   
   async def review_plan(plan) -> bool:
       return True  # your async check here
   
   approval = ApprovalCallback(approve_fn=review_plan)
   
   async def main():
       plan = Plan(name="Weekly report")
       # Reached from inside a running loop — no new-loop crash.
       approved = approval(plan)
       print("approved" if approved else "rejected")
   
   asyncio.run(main())

How It Works

The sync __call__ detects an async approve_fn and forwards it through the package’s shared safe bridge run_coroutine_from_any_context, which detects a running loop and schedules the coroutine on it instead of spinning up a fresh loop.

FastAPI async route

Embed PraisonAI in a FastAPI async endpoint and the plan-approval prompt no longer raises RuntimeError when the plan is presented.

from fastapi import FastAPI
from praisonaiagents import Agent, PlanningConfig
from praisonaiagents.planning import ApprovalCallback

app = FastAPI()

async def review_plan(plan) -> bool:
    # e.g. call a Slack / email / DB check
    return await my_async_approval_check(plan)

approval = ApprovalCallback(approve_fn=review_plan)

@app.post("/plan")
async def run_plan():
    agent = Agent(
        name="Planner",
        instructions="Solve step by step",
        planning=PlanningConfig(auto_approve=False),
    )
    # Works inside the route's running loop — no new-loop crash.
    return await agent.astart("Plan the weekly report")

What the user experiences

• A user hits a FastAPI endpoint that runs await agent.astart("...").
• The planning phase produces a plan; ApprovalCallback is invoked.
• The user-supplied async approve_fn runs on the caller’s loop (no new-loop crash).
• The plan proceeds if approved, or halts if the callback returns False.

Common Patterns

from praisonaiagents.planning import ApprovalCallback

async def review_plan(plan) -> bool:
    return await my_async_approval_check(plan)

approval = ApprovalCallback(approve_fn=review_plan)

from praisonaiagents.planning import ApprovalCallback

def review_plan(plan) -> bool:
    return len(plan.steps) <= 5

approval = ApprovalCallback(approve_fn=review_plan)

from praisonaiagents.planning import ApprovalCallback

approval = ApprovalCallback(auto_approve=True)

Best Practices

Related