What Are AI Agents and Why They Matter for Business Automation
AI agents are autonomous software programs that use large language models (LLMs) to perceive their environment, make decisions, and take actions to achieve specific goals. Unlike traditional automation scripts that follow rigid if-then rules, AI agents can reason, adapt, and handle ambiguous situations—making them perfect for automating complex business workflows that previously required human judgment.
When I first started exploring AI agents, I was drowning in operational tasks across my SaaS business: customer support tickets, lead qualification, content scheduling, invoice follow-ups, and data entry. Each task required context switching that cost me roughly 40 hours per week. By the end of this journey, I had reduced my active workload to under 5 hours per week while actually improving response quality and consistency.
Here's what makes AI agents transformative for business automation:
- Tool use: Agents can call APIs, query databases, send emails, and update CRM records—bridging the gap between reasoning and real-world action
- Memory and persistence: They maintain context across conversations and tasks, learning preferences and patterns over time
- Planning and decomposition: Complex goals are broken into sequential steps, with the agent adjusting its plan when obstacles arise
- Guardrails and safety: You define boundaries—what the agent can and cannot do—preventing costly mistakes
- 24/7 operation: Agents never sleep, never get distracted, and scale horizontally across hundreds of concurrent tasks
Architecture Overview: The Agent Orchestration Layer
Before diving into code, you need to understand the architectural pattern that makes this work. I use a centralized Agent Orchestrator that manages multiple specialized agents, each with its own tool set and system prompt. The orchestrator routes incoming tasks, maintains a shared memory store, and handles escalation when agents encounter problems they cannot solve.
Here's the high-level architecture I settled on after multiple iterations:
┌─────────────────────────────────────────────┐
│ Task Ingestion Layer │
│ (Webhooks, Email, Slack, API, Scheduled) │
└──────────────────┬──────────────────────────┘
│
┌──────────────────▼──────────────────────────┐
│ Agent Orchestrator (FastAPI) │
│ - Routes tasks to specialized agents │
│ - Maintains conversation/thread memory │
│ - Handles escalation & human-in-the-loop │
└──┬──────────────┬──────────────┬─────────────┘
│ │ │
┌──▼──┐ ┌────▼────┐ ┌────▼────┐
│Support│ │ Sales │ │ Ops │ ...more
│Agent │ │ Agent │ │ Agent │
└──┬───┘ └────┬────┘ └───┬─────┘
│ │ │
┌──▼─────────────▼──────────────▼─────────────┐
│ Tool Execution Layer │
│ Stripe • Gmail • Slack • PostgreSQL │
│ Zendesk • Calendly • Twilio • Vector DB │
└─────────────────────────────────────────────┘
Each agent is essentially a system prompt + a curated set of tools + a reasoning loop. The orchestrator doesn't need to be complex—it's mostly a routing layer with some state management. The real intelligence lives inside each specialized agent.
Step 1: Setting Up Your Agent Framework
I evaluated several frameworks before settling on LangChain's agent runtime combined with a custom FastAPI wrapper. Here's the foundation every agent in my system shares. Create a file called agent_core.py:
# agent_core.py
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.tools import BaseTool
from langchain.schema import SystemMessage, HumanMessage
from pydantic import BaseModel, Field
from typing import List, Dict, Any, Optional
import json
import asyncio
class AgentConfig(BaseModel):
"""Configuration for a single AI agent"""
name: str
role_description: str
tools: List[BaseTool]
model_name: str = "gpt-4-turbo-preview"
temperature: float = 0.2
max_iterations: int = 8
require_approval_for: List[str] = Field(default_factory=list)
# Actions that need human approval before execution
memory_keys: List[str] = Field(default_factory=lambda: ["chat_history"])
class BaseBusinessAgent:
"""Foundation class that all specialized agents inherit from"""
def __init__(self, config: AgentConfig, memory_store=None):
self.config = config
self.memory_store = memory_store or InMemoryConversationStore()
self.llm = ChatOpenAI(
model=config.model_name,
temperature=config.temperature,
)
self._setup_agent()
def _setup_agent(self):
system_prompt = f"""You are {self.config.name}, an AI agent responsible for:
{self.config.role_description}
CRITICAL RULES:
1. Before executing any action marked as requiring approval,
you MUST pause and request explicit human confirmation.
Actions requiring approval: {', '.join(self.config.require_approval_for)}
2. If you encounter an error after 3 attempts, escalate to a human.
3. Always log your reasoning before taking actions.
4. Never hallucinate data—if you're unsure, ask for clarification.
5. Maintain professional, empathetic tone in all communications.
You have access to specialized tools. Use them appropriately.
"""
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
agent = create_openai_tools_agent(
llm=self.llm,
tools=self.config.tools,
prompt=prompt
)
self.executor = AgentExecutor(
agent=agent,
tools=self.config.tools,
max_iterations=self.config.max_iterations,
verbose=True,
handle_parsing_errors=True,
return_intermediate_steps=True,
)
async def run(self, task_input: str, context: Dict[str, Any] = None) -> Dict[str, Any]:
"""Execute a task and return structured result"""
memory = self.memory_store.get_context(
thread_id=context.get("thread_id", "default")
)
enriched_input = task_input
if context:
enriched_input = f"""Context information:
{json.dumps(context, indent=2)}
User task: {task_input}"""
result = await self.executor.ainvoke({
"input": enriched_input,
"chat_history": memory
})
# Update memory
self.memory_store.save_context(
thread_id=context.get("thread_id", "default"),
messages=[
HumanMessage(content=task_input),
SystemMessage(content=result.get("output", ""))
]
)
return {
"success": True,
"output": result["output"],
"intermediate_steps": result.get("intermediate_steps", []),
"agent": self.config.name
}
The BaseBusinessAgent class encapsulates the reasoning loop, memory management, and safety guardrails. Every specialized agent you build later will extend this foundation. The approval mechanism is critical—it prevents agents from autonomously executing high-risk actions like issuing refunds or sending contracts without your review.
Step 2: Building Your First Specialized Agent—Customer Support
Customer support was my biggest time sink, so it became my first agent. This agent handles ticket triage, responds to common questions using knowledge base articles, and escalates complex issues. Here's the complete implementation:
# agents/support_agent.py
from agent_core import BaseBusinessAgent, AgentConfig
from langchain.tools import tool, BaseTool
from typing import Optional
import stripe
import requests
from datetime import datetime, timedelta
# ---- Custom Tools for Support Agent ----
class LookupCustomerTool(BaseTool):
name: str = "lookup_customer"
description: str = "Look up customer details by email or customer ID. Returns subscription status, plan, and history."
def _run(self, identifier: str) -> str:
"""Query Stripe for customer information"""
try:
if "@" in identifier:
customers = stripe.Customer.list(email=identifier, limit=1)
else:
customers = stripe.Customer.list(query=f"id:'{identifier}'", limit=1)
if not customers.data:
return f"No customer found for: {identifier}"
customer = customers.data[0]
subscriptions = stripe.Subscription.list(customer=customer.id, limit=3)
sub_info = []
for sub in subscriptions.data:
sub_info.append({
"status": sub.status,
"plan": sub.items.data[0].price.id if sub.items.data else "unknown",
"current_period_end": datetime.fromtimestamp(sub.current_period_end).isoformat(),
"created": datetime.fromtimestamp(sub.created).isoformat()
})
return f"""Customer: {customer.email}
Name: {customer.name or 'N/A'}
Created: {datetime.fromtimestamp(customer.created).isoformat()}
Balance: ${customer.balance / 100:.2f}
Subscriptions: {json.dumps(sub_info, indent=2)}"""
except Exception as e:
return f"Error looking up customer: {str(e)}"
class SearchKnowledgeBase(BaseTool):
name: str = "search_knowledge_base"
description: str = "Search the internal knowledge base for articles matching a query. Returns relevant article excerpts."
def _run(self, query: str) -> str:
"""Query vector database for relevant knowledge articles"""
# In production, this queries Pinecone/Chroma with embeddings
kb_endpoint = "https://kb-api.yourcompany.com/search"
response = requests.post(kb_endpoint, json={"query": query, "top_k": 3})
if response.status_code != 200:
return "Knowledge base temporarily unavailable"
articles = response.json().get("results", [])
if not articles:
return f"No relevant articles found for: {query}"
formatted = []
for art in articles:
formatted.append(f"## {art['title']}\n{art['excerpt']}\nRelevance: {art['score']:.2f}")
return "\n\n".join(formatted)
class CreateSupportTicket(BaseTool):
name: str = "create_support_ticket"
description: str = "Create a new support ticket in Zendesk when an issue needs human attention. REQUIRES APPROVAL."
def _run(self, customer_email: str, subject: str, body: str, priority: str = "normal") -> str:
"""Create Zendesk ticket—this tool requires human approval"""
# The approval check happens at the orchestrator level
ticket_data = {
"ticket": {
"requester": {"email": customer_email},
"subject": subject,
"comment": {"body": body},
"priority": priority,
"tags": ["ai-agent-created", "requires-review"]
}
}
# This would actually post to Zendesk API
return f"Ticket would be created for {customer_email}: {subject} (priority: {priority})\nBody: {body[:200]}..."
# ---- Assembling the Support Agent ----
support_agent_tools = [
LookupCustomerTool(),
SearchKnowledgeBase(),
CreateSupportTicket()
]
support_agent_config = AgentConfig(
name="SupportAgent",
role_description="""You handle all customer support inquiries for a SaaS platform.
Your responsibilities:
1. Triage incoming support requests by urgency
2. Answer common questions using the knowledge base
3. Look up customer account details when needed
4. For billing issues, check Stripe subscription status
5. Create support tickets ONLY when the issue cannot be resolved automatically
6. For refund requests over $100, always escalate to human review""",
tools=support_agent_tools,
require_approval_for=["create_support_ticket"],
temperature=0.1 # Low temperature for consistent support responses
)
support_agent = BaseBusinessAgent(support_agent_config)
The key design decision here is tool granularity. Each tool does exactly one thing well. The agent's system prompt determines when and how to combine them. Notice that CreateSupportTicket is marked as requiring approval—the orchestrator will intercept this and ping you on Slack before proceeding.
Step 3: The Sales Agent—Lead Qualification on Autopilot
My sales agent handles inbound leads from the website, qualifies them against ideal customer profile criteria, schedules meetings, and drafts follow-up emails. This replaced a full-time SDR function:
# agents/sales_agent.py
from agent_core import BaseBusinessAgent, AgentConfig
from langchain.tools import BaseTool
from typing import Optional, Dict
import json
import requests
import os
class EnrichLeadData(BaseTool):
name: str = "enrich_lead_data"
description: str = "Enrich a lead with company data from Clearbit. Provide an email domain or company name."
def _run(self, company_domain: str) -> str:
clearbit_key = os.environ.get("CLEARBIT_API_KEY")
url = f"https://company.clearbit.com/v2/companies/domain/{company_domain}"
headers = {"Authorization": f"Bearer {clearbit_key}"}
response = requests.get(url, headers=headers)
if response.status_code != 200:
return f"Could not enrich: {company_domain}"
data = response.json()
return f"""Company: {data.get('name', 'Unknown')}
Industry: {data.get('category', {}).get('industry', 'Unknown')}
Size: {data.get('metrics', {}).get('employees', 'Unknown')} employees
Revenue: {data.get('metrics', {}).get('annualRevenue', 'Unknown')}
Tech Stack: {', '.join(data.get('tech', []))}
Description: {data.get('description', 'N/A')[:300]}"""
class ScoreLead(BaseTool):
name: str = "score_lead"
description: str = "Score a lead based on ICP criteria. Returns score and qualification level."
ICP_CRITERIA = {
"industry_match": ["Technology", "SaaS", "Fintech", "Healthcare IT"],
"min_employees": 50,
"max_employees": 2000,
"decision_maker_title_keywords": ["VP", "Director", "Head", "CTO", "CIO", "CEO"],
"budget_indicators": ["evaluation", "pilot", "RFP", "procurement"]
}
def _run(self, lead_data_json: str) -> str:
lead = json.loads(lead_data_json)
score = 0
reasons = []
industry = lead.get("industry", "")
if any(ind.lower() in industry.lower() for ind in self.ICP_CRITERIA["industry_match"]):
score += 30
reasons.append("Industry match (+30)")
employees = lead.get("employees", 0)
if self.ICP_CRITERIA["min_employees"] <= employees <= self.ICP_CRITERIA["max_employees"]:
score += 25
reasons.append("Company size in range (+25)")
title = lead.get("title", "")
if any(kw.lower() in title.lower() for kw in self.ICP_CRITERIA["decision_maker_title_keywords"]):
score += 25
reasons.append("Decision-maker title (+25)")
source = lead.get("source", "")
if source in ["demo_request", "pricing_page", "contact_sales"]:
score += 15
reasons.append("High-intent source (+15)")
if score >= 70:
qualification = "HOT LEAD - Schedule immediately"
elif score >= 50:
qualification = "WARM LEAD - Nurture sequence"
else:
qualification = "COLD LEAD - Long-term nurture"
return f"""Score: {score}/100
Qualification: {qualification}
Reasons: {', '.join(reasons)}"""
class ScheduleMeeting(BaseTool):
name: str = "schedule_meeting"
description: str = "Check calendar availability and propose meeting times. REQUIRES APPROVAL."
def _run(self, lead_email: str, preferred_time: str = "") -> str:
# Query Calendly or Google Calendar API
# This is a simplified example
return f"Calendar check for meetings with {lead_email}: Available slots found on Tue 2pm, Wed 10am, Thu 3pm."
# Sales agent assembly
sales_agent_tools = [EnrichLeadData(), ScoreLead(), ScheduleMeeting()]
sales_agent_config = AgentConfig(
name="SalesAgent",
role_description="""You are an inbound sales qualification agent.
Process:
1. Enrich incoming leads with company data
2. Score against ICP criteria
3. For HOT leads (score >= 70), propose meeting times
4. For WARM leads, suggest nurture email content
5. Never over-promise product capabilities
6. Always be truthful about limitations""",
tools=sales_agent_tools,
require_approval_for=["schedule_meeting"],
temperature=0.3
)
sales_agent = BaseBusinessAgent(sales_agent_config)
The scoring logic is transparent and auditable—you can see exactly why a lead received a particular score. This is crucial for business automation because you need to trust the agent's decisions. The ICP criteria are configurable constants that you can adjust as your ideal customer profile evolves.
Step 4: The Orchestrator—Tying Everything Together
The orchestrator is the brain stem of the operation. It receives tasks from various channels, routes them to the correct agent, enforces approval workflows, and maintains a global view of all agent activity. Here's the production-ready implementation:
# orchestrator.py
from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel, EmailStr
from typing import Dict, Any, Optional, List
from enum import Enum
import asyncio
import hashlib
import json
from datetime import datetime
import redis
import smtplib
from email.mime.text import MIMEText
app = FastAPI(title="Agent Orchestrator")
# Initialize Redis for shared memory
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
class TaskType(str, Enum):
SUPPORT_TICKET = "support_ticket"
LEAD_QUALIFICATION = "lead_qualification"
INVOICE_FOLLOWUP = "invoice_followup"
CONTENT_SCHEDULE = "content_schedule"
DATA_SYNC = "data_sync"
class TaskRequest(BaseModel):
task_type: TaskType
payload: Dict[str, Any]
thread_id: Optional[str] = None
priority: str = "normal"
class PendingApproval(BaseModel):
approval_id: str
agent: str
action: str
arguments: Dict[str, Any]
reasoning: str
timestamp: datetime
# Store pending approvals
pending_approvals: Dict[str, PendingApproval] = {}
# Agent registry
agent_registry = {
TaskType.SUPPORT_TICKET: support_agent, # from step 2
TaskType.LEAD_QUALIFICATION: sales_agent, # from step 3
# Additional agents registered here
}
def extract_approval_requests(result: Dict[str, Any]) -> List[Dict]:
"""Parse agent output for approval requests"""
approvals = []
for step in result.get("intermediate_steps", []):
action = step[0] # AgentAction
if hasattr(action, 'tool') and action.tool in ["create_support_ticket", "schedule_meeting"]:
approvals.append({
"tool": action.tool,
"arguments": action.tool_input,
"reasoning": step[1] if len(step) > 1 else "No reasoning provided"
})
return approvals
async def send_approval_notification(approval: PendingApproval):
"""Send Slack/email notification for human approval"""
msg = MIMEText(f"""Agent: {approval.agent}
Action: {approval.action}
Arguments: {json.dumps(approval.arguments, indent=2)}
Reasoning: {approval.reasoning}
Approve: https://dashboard.yourcompany.com/approve/{approval.approval_id}
Reject: https://dashboard.yourcompany.com/reject/{approval.approval_id}""")
msg['Subject'] = f"🔔 Agent Approval Needed: {approval.action}"
msg['From'] = "orchestrator@yourcompany.com"
msg['To'] = "you@yourcompany.com"
# Actually send via SMTP or Slack webhook
print(f"[APPROVAL] {approval.approval_id}: {approval.action}")
@app.post("/task")
async def handle_task(request: TaskRequest, background_tasks: BackgroundTasks):
"""Main entry point for all agent tasks"""
thread_id = request.thread_id or hashlib.md5(
json.dumps(request.payload).encode()
).hexdigest()[:16]
# Get agent for this task type
agent = agent_registry.get(request.task_type)
if not agent:
raise HTTPException(400, f"No agent for task type: {request.task_type}")
# Execute agent
result = await agent.run(
task_input=json.dumps(request.payload),
context={"thread_id": thread_id, "task_type": request.task_type}
)
# Check for approval requests in agent's intermediate steps
approvals = extract_approval_requests(result)
if approvals:
approval_id = f"appr-{datetime.now().timestamp()}"
pending = PendingApproval(
approval_id=approval_id,
agent=agent.config.name,
action=approvals[0]["tool"],
arguments=approvals[0]["arguments"],
reasoning=approvals[0]["reasoning"],
timestamp=datetime.now()
)
pending_approvals[approval_id] = pending
# Notify human asynchronously
background_tasks.add_task(send_approval_notification, pending)
return {
"status": "pending_approval",
"approval_id": approval_id,
"agent_output": result["output"],
"requires_action": approvals[0]["tool"]
}
return {
"status": "completed",
"output": result["output"],
"agent": result["agent"]
}
@app.post("/approve/{approval_id}")
async def approve_action(approval_id: str):
"""Human approves a pending agent action"""
pending = pending_approvals.pop(approval_id, None)
if not pending:
raise HTTPException(404, "Approval not found")
# Re-execute the specific action with approval flag
# In production, this would resume the agent from the checkpoint
return {
"status": "approved",
"action": pending.action,
"message": f"Action {pending.action} has been approved and will execute"
}
@app.get("/dashboard")
async def agent_dashboard():
"""Overview of all agent activity"""
return {
"pending_approvals": len(pending_approvals),
"tasks_completed_today": int(redis_client.get("tasks:today") or 0),
"active_threads": redis_client.scard("active_threads"),
"agents": list(agent_registry.keys()),
"uptime": "99.9%"
}
# Health check and monitoring
@app.get("/health")
async def health():
return {"status": "operational", "agents": len(agent_registry)}
The orchestrator solves the hardest problem in AI agent systems: trustworthy execution of consequential actions. By intercepting tool calls that match the require_approval_for list, it creates a human-in-the-loop checkpoint. The agent's reasoning is preserved, so when you approve, the action executes exactly as the agent intended—but only after you've verified it makes sense.
Step 5: Operations Agent—The Silent Workhorse
While support and sales agents are customer-facing, the operations agent handles backend tasks that keep the business running. This agent monitors Stripe for failed payments, sends reminder emails, updates CRM records, and generates weekly reports:
# agents/ops_agent.py
from agent_core import BaseBusinessAgent, AgentConfig
from langchain.tools import BaseTool
from datetime import datetime, timedelta
import stripe
import json
class CheckFailedPayments(BaseTool):
name: str = "check_failed_payments"
description: str = "Check Stripe for recent failed payments and return list of affected customers"
def _run(self, hours_lookback: int = 24) -> str:
cutoff = datetime.now() - timedelta(hours=hours_lookback)
# Query Stripe for failed payment intents
failed = stripe.PaymentIntent.list(
created={"gte": int(cutoff.timestamp())},
status="failed",
limit=50
)
results = []
for pi in failed.data:
customer = stripe.Customer.retrieve(pi.customer) if pi.customer else None
results.append({
"customer_email": customer.email if customer else "unknown",
"amount": pi.amount / 100,
"failure_reason": pi.last_payment_error.message if pi.last_payment_error else "unknown",
"created": datetime.fromtimestamp(pi.created).isoformat()
})
return json.dumps(results, indent=2) if results else "No failed payments in the last 24 hours"
class SendReminderEmail(BaseTool):
name: str = "send_reminder_email"
description: str = "Send a payment reminder email to a customer. REQUIRES APPROVAL for amounts over $500."
def _run(self, customer_email: str, amount: float, days_overdue: int) -> str:
# Template the email
email_body = f"""Subject: Gentle Reminder: Payment Due for ${amount:.2f}
Dear Customer,
Our records show that a payment of ${amount:.2f} is now {days_overdue} days overdue.
To avoid any service interruption, please update your payment method at:
https://billing.yourcompany.com/update-payment
If you've already made this payment, please disregard this message.
Need help? Reply to this email or chat with us at https://yourcompany.com/support
Best regards,
The Billing Team"""
return f"Email prepared for {customer_email}\n---\n{email_body}\n---\nReady to send upon approval."
class GenerateWeeklyReport(BaseTool):
name: str = "generate_weekly_report"
description: str = "Generate a weekly business metrics report pulling from Stripe, analytics, and CRM"
def _run(self, week_ending: str) -> str:
# Aggregate data from multiple sources
# Stripe MRR, new customers, churn
stripe_data = {
"mrr": stripe.Subscription.list(status="active", limit=1000).data,
"new_customers": stripe.Customer.list(
created={"gte": int((datetime.now() - timedelta(days=7)).timestamp())}
).data
}
report = f"""Weekly Business Report - Week ending {week_ending}
Revenue Metrics:
- Active Subscriptions: {len(stripe_data['mrr'])}
- New Customers (7d): {len(stripe_data['new_customers'])}
- Estimated MRR Change: ${sum(s.items.data[0].price.unit_amount * s.items.data[0].quantity / 100 for s in stripe_data['mrr'] if s.items.data):.2f}
Key Actions Taken This Week:
- Failed payments handled: {self._get_failed_count()}
- Support tickets resolved: {self._get_resolved_tickets()}
- Leads qualified: {self._get_qualified_leads()}
Generated by OpsAgent at {datetime.now().isoformat()}
"""
return report
ops_agent_tools = [CheckFailedPayments(), SendReminderEmail(), GenerateWeeklyReport()]
ops_agent_config = AgentConfig(
name="OpsAgent",
role_description="""You are the operations agent handling backend business processes.
Daily tasks:
1. Check for failed payments every morning at 9am
2. Send reminders for payments 3+ days overdue
3. Generate weekly reports every Monday
4. Monitor for anomalies (spike in churn, unusual refund patterns)
5. Update CRM with latest subscription data""",
tools=ops_agent_tools,
require_approval_for=["send_reminder_email"],
temperature=0.1
)
ops_agent = BaseBusinessAgent(ops_agent_config)
Step 6: Scheduled Agent Execution with Celery
Agents need to run on schedules—checking failed payments every morning, generating reports on Mondays, etc. I use Celery with Redis as the broker for reliable scheduled execution:
# scheduler.py
from celery import Celery
from celery.schedules import crontab
from orchestrator import TaskRequest, TaskType
import asyncio
import requests
import os
celery_app = Celery('agent_scheduler', broker='redis://localhost:6379/1')
ORCHESTRATOR_URL = os.environ.get("ORCHESTRATOR_URL", "http://localhost:8000")
@celery_app.task
def execute_agent_task(task_type: str, payload: dict):
"""Generic task executor that calls the orchestrator"""
response = requests.post(
f"{ORCHESTRATOR_URL}/task",
json={
"task_type": task_type,
"payload": payload,
"priority": "normal"
}
)
return response.json()
# ---- Scheduled Tasks ----
@celery_app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
"""Register all recurring agent tasks"""
# Check failed payments every morning at 9am
sender.add_periodic_task(
crontab(hour=9, minute=0),
check_failed_payments.s(),
name="check-failed-payments-daily"
)
# Generate weekly report every Monday at 8am
sender.add_periodic_task(
crontab(hour=8, minute=0, day_of_week=1),
generate_weekly_report.s(),
name="generate-weekly-report"
)
# Sync CRM data every 2 hours
sender.add_periodic_task(
crontab(minute=0, hour='*/2'),
sync_crm_data.s(),
name="sync-crm-every-2h"
)
# Inbound lead processing every 30 minutes
sender.add_periodic_task(
crontab(minute='*/30'),
process_inbound_leads.s(),
name="process-leads-30min"
)
@celery_app.task
def check_failed_payments():
return execute_agent_task("invoice_followup", {
"action": "check_and_remind",
"max_days_overdue": 7
})
@celery_app.task
def generate_weekly_report():
return execute_agent_task("data_sync", {
"action": "weekly_report",
"send_to": ["ceo@yourcompany.com", "ops@yourcompany.com"]
})
@celery_app.task
def sync_crm_data():
return execute_agent_task("data_sync", {
"action": "sync_subscriptions_to_hubspot"
})
@celery_app.task
def process_inbound_leads():
# Fetch leads from a queue or database
return execute_agent_task("lead_qualification", {
"action": "process_queue",
"source": "website_forms"
})
This scheduler pattern is deliberately simple. Each scheduled task is just a thin wrapper that calls the orchestrator's /task endpoint. The orchestrator handles routing, agent selection, and approval workflows. You can monitor everything from the /dashboard endpoint.
Step 7: Webhook Integration—Connecting External Services
To make the system truly autonomous, you need inbound webhooks from Stripe, your CRM, email, and other services. Here's how to wire them into the orchestrator:
# webhooks.py - Add these routes to orchestrator.py
from fastapi import Request
import stripe
import hmac
import hashlib
@app.post("/webhook/stripe")
async def stripe_webhook(request: Request):
"""Handle Stripe events and route to appropriate agent"""
payload = await request.body()
sig_header = request.headers.get("stripe-signature")
try:
event = stripe.Webhook.construct_event(
payload, sig_header, os.environ["STRIPE_WEBHOOK_SECRET"]
)
except stripe.error.SignatureVerificationError:
raise HTTPException(400, "Invalid signature")
event_type = event.type
data = event.data.object
# Route events to agents
if event_type == "payment_intent.payment_failed":
task = TaskRequest(
task_type=TaskType.INVOICE_FOLLOWUP,
payload={
"event": "payment_failed",
"customer_id": data.get("customer"),
"amount": data.get("amount") / 100,
"failure_reason": data.get("last_payment_error", {}).get("message", "unknown")
}
)
await handle_task(task, BackgroundTasks())
elif event_type == "customer.subscription.deleted":
task = TaskRequest(
task_type=TaskType.DATA_SYNC,
payload={
"event": "subscription_cancelled",
"customer_id": data.get("customer"),
"cancel_reason": data.get("cancellation_details", {}).get("reason", "unknown")
}
)
await handle_task(task, BackgroundTasks())
return {"status": "processed"}
@app.post("/webhook/email/inbound")
async def inbound_email_webhook(request: Request):
"""Parse inbound emails and route to support or sales agent"""
body = await request.json()
from_address = body.get("from")
subject = body.get("subject", "")
text_content = body.get("text", "")
# Classify intent from email content
intent = classify_email_intent(subject, text_content)
if intent == "support":
task_type = TaskType.SUPPORT_TICKET
payload = {
"customer_email": from_address,
"subject": subject,
"body": text_content,
"source": "email"
}
elif intent == "sales":
task_type = TaskType.LEAD_QUALIFICATION
payload = {
"email": from_address,
"subject": subject,
"body": text_content,