Trigger-Action Agent Patterns
Event detection → decision → action: the core automation loop.
Trigger-Action Agent Patterns is a free AI Agents lesson on CoddyKit — lesson 1 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Agents learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.
What Is a Trigger-Action Agent?
A trigger-action agent watches for events and responds with actions. The three-part loop is: detect event → LLM decides action → execute action.
Examples: email received → summarize and reply; file uploaded → validate and process; 9am every day → generate daily report.
Types of Triggers
Triggers fall into three categories:
- Event-based: webhook fires when email arrives or file uploads
- Time-based: cron schedule runs agent at fixed intervals
- Polling-based: agent checks an API repeatedly for new data
Choosing the right trigger type determines your agent's latency and resource usage.
The Detect Phase
Detection means receiving or recognizing an event. For webhooks, your server receives a POST request. For polling, your agent queries an API and compares results to last-seen state.
import json
def detect_new_email(current_emails, last_seen_id):
new_emails = [
e for e in current_emails
if e['id'] > last_seen_id
]
return new_emails
# Simulate detection
current = [{'id': 3, 'subject': 'Meeting'}, {'id': 4, 'subject': 'Invoice'}]
new = detect_new_email(current, last_seen_id=2)
print('New emails:', [e['subject'] for e in new])The Decide Phase
After detecting an event, the agent sends context to an LLM and asks what action to take. The LLM either selects a tool or returns a direct response.
import openai
client = openai.OpenAI(api_key='sk-...')
def decide_action(event_description):
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[
{'role': 'system', 'content': 'You are an automation agent. Decide what action to take for the event.'},
{'role': 'user', 'content': f'Event: {event_description}'}
],
tools=[
{'type': 'function', 'function': {'name': 'send_reply', 'description': 'Reply to email', 'parameters': {'type': 'object', 'properties': {'message': {'type': 'string'}}, 'required': ['message']}}}
]
)
return response.choices[0].message
result = decide_action('New email: Invoice for $500 from supplier')
print(result)The Execute Phase
Execution runs the chosen action. This might call an API, write a file, send a message, or trigger another workflow. Always handle errors and log outcomes.
import logging
import sys
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
logger = logging.getLogger('agent')
def execute_action(action_name, params):
try:
if action_name == 'send_reply':
# In real code, call Gmail API here
logger.info(f'Sending reply: {params["message"]}')
return {'status': 'success'}
elif action_name == 'create_task':
logger.info(f'Creating task: {params["title"]}')
return {'status': 'success'}
else:
raise ValueError(f'Unknown action: {action_name}')
except Exception as e:
logger.error(f'Action failed: {e}')
return {'status': 'error', 'message': str(e)}
if __name__ == '__main__':
result = execute_action('send_reply', {'message': 'Thanks for reaching out!'})
print('Result:', result)
State Machine Model
A state machine is a powerful model for automation agents. States might be: IDLE, DETECTING, DECIDING, EXECUTING, ERROR. Transitions happen on events or conditions.
State machines make agent behavior predictable and easier to debug.
from enum import Enum
class AgentState(Enum):
IDLE = 'idle'
DETECTING = 'detecting'
DECIDING = 'deciding'
EXECUTING = 'executing'
ERROR = 'error'
class AutomationAgent:
def __init__(self):
self.state = AgentState.IDLE
def transition(self, new_state):
print(f'State: {self.state.value} -> {new_state.value}')
self.state = new_state
def run_cycle(self, event=None):
self.transition(AgentState.DETECTING)
if event:
self.transition(AgentState.DECIDING)
self.transition(AgentState.EXECUTING)
self.transition(AgentState.IDLE)
agent = AutomationAgent()
agent.run_cycle(event={'type': 'email', 'subject': 'Test'})Email Received Trigger Pattern
Gmail push notifications use Pub/Sub. When a new email arrives, Google publishes to your topic. Your agent receives a webhook, fetches the email, processes it.
from fastapi import FastAPI, Request
import base64, json
app = FastAPI()
@app.post('/gmail-push')
async def gmail_push(request: Request):
body = await request.json()
# Decode Pub/Sub message
message = body.get('message', {})
data = base64.b64decode(message.get('data', '')).decode('utf-8')
notification = json.loads(data)
email_address = notification.get('emailAddress')
history_id = notification.get('historyId')
print(f'New email for {email_address}, historyId: {history_id}')
# Fetch email details and run agent here
return {'status': 'ok'}File Upload Trigger Pattern
S3 event notifications or local filesystem watchers can trigger agents when files appear. The watchdog library watches directories for new files.
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time
class UploadHandler(FileSystemEventHandler):
def on_created(self, event):
if event.is_directory:
return
print(f'New file detected: {event.src_path}')
self.process_file(event.src_path)
def process_file(self, filepath):
# Run agent logic on new file
print(f'Processing: {filepath}')
observer = Observer()
handler = UploadHandler()
observer.schedule(handler, path='/tmp/uploads/', recursive=False)
observer.start()
try:
time.sleep(30) # Watch for 30 seconds
finally:
observer.stop()
observer.join()Time-Based Trigger Pattern
Time-based triggers fire agents on a schedule. Use APScheduler for in-process scheduling or a system cron job for process-level scheduling.
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
scheduler = BlockingScheduler()
def daily_report_agent():
print(f'Daily report running at {datetime.now()}')
# Fetch data, call LLM, send report
pass
def hourly_check_agent():
print(f'Hourly check at {datetime.now()}')
pass
# Run at 8am every day
scheduler.add_job(daily_report_agent, 'cron', hour=8, minute=0)
# Run every 30 minutes
scheduler.add_job(hourly_check_agent, 'interval', minutes=30)
print('Scheduler started')
scheduler.start()Idempotent Action Execution
Automation agents must be idempotent: running the same action twice should not cause duplicate effects. Use idempotency keys and check-before-act patterns.
import hashlib
processed_events = set() # In production, use Redis or DB
def compute_event_id(event):
content = f"{event['type']}:{event['source_id']}:{event['timestamp']}"
return hashlib.sha256(content.encode()).hexdigest()[:16]
def handle_event_idempotent(event):
event_id = compute_event_id(event)
if event_id in processed_events:
print(f'Skipping duplicate event: {event_id}')
return {'status': 'duplicate', 'event_id': event_id}
# Process event
print(f'Processing event: {event_id}')
processed_events.add(event_id)
return {'status': 'processed', 'event_id': event_id}
# Simulate duplicate event
event = {'type': 'email', 'source_id': 'abc123', 'timestamp': '2024-01-01T09:00:00'}
print(handle_event_idempotent(event))
print(handle_event_idempotent(event)) # Duplicate - skippedError States and Recovery
Robust agents handle failures gracefully. When execution fails, the agent can retry with exponential backoff, alert a human, or move to a dead-letter queue for manual review.
import time
def execute_with_retry(action_fn, max_retries=3):
for attempt in range(max_retries):
try:
result = action_fn()
print(f'Success on attempt {attempt + 1}')
return result
except Exception as e:
wait = 2 ** attempt # Exponential backoff: 1s, 2s, 4s
print(f'Attempt {attempt + 1} failed: {e}. Retrying in {wait}s')
if attempt < max_retries - 1:
time.sleep(wait)
else:
print('All retries exhausted. Moving to dead-letter queue.')
raise
# Example usage
call_count = [0]
def flaky_action():
call_count[0] += 1
if call_count[0] < 3:
raise ConnectionError('Service unavailable')
return 'Done'
execute_with_retry(flaky_action)Knowledge Check: Trigger-Action Patterns
Test your understanding of trigger-action agent patterns.
Putting It Together
A complete trigger-action agent combines all the pieces: a trigger source (email, file, timer), a detection layer, LLM-based decision making, idempotent execution, retry logic, and state tracking.
Start simple: one trigger type, one action. Add complexity incrementally as you gain confidence in the agent's behavior.
Frequently asked questions
Is the “Trigger-Action Agent Patterns” lesson free?
Yes — the full text of “Trigger-Action Agent Patterns” is free to read here on the web, and the AI Agents course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Agents course, upgrade to CoddyKit PRO.
What will I learn in “Trigger-Action Agent Patterns”?
Event detection → decision → action: the core automation loop. You practise AI Agents with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.
Do I need any experience to start AI Agents?
No prior experience is required. AI Agents on CoddyKit is structured for beginners through advanced learners; this is — lesson 1 of 4, so you can start here or from the beginning and move at your own pace.
How long does the “Trigger-Action Agent Patterns” lesson take?
Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.
Can I write and run code in this AI Agents lesson?
Yes. Every AI Agents lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.
All lessons in this course
- Trigger-Action Agent Patterns
- Connecting Agents to Webhooks
- Scheduling and Cron-Based Agents
- Building a Multi-App Automation Pipeline