AutoGen’s human-in-the-loop capability becomes practical when you route approval requests through Slack instead of blocking on console input. This guide walks through a complete autogen slack human approval integration that handles timeouts, retries, and audit logging — the pieces most tutorials skip. You’ll end up with a pattern you can drop into any multi-agent workflow.
Why Slack for human-in-the-loop
Console-based input() works for local scripts but fails in production: no visibility for teammates, no audit trail, no mobile access, and no way to escalate when the primary approver is unavailable. Slack solves all four. Your agents post structured approval requests to a channel, stakeholders click buttons, and the workflow resumes — or escalates — automatically.
The tradeoff is latency. A Slack round-trip adds 500 ms to 2 seconds versus a local call. For most business processes (code review, deployment gates, expense approval) that’s negligible. For high-frequency trading or robotics control, it’s not. Choose accordingly.
Architecture overview
┌─────────────┐ ┌──────────────┐ ┌─────────────┐
│ AutoGen │────▶│ Approval │────▶│ Slack │
│ Agent │ │ Manager │ │ Workspace │
└─────────────┘ └──────────────┘ └─────────────┘
│ │
▼ ▼
┌──────────────┐ ┌─────────────┐
│ State Store │ │ Interactive│
│ (Redis/DB) │ │ Components │
└──────────────┘ └─────────────┘The approval manager is a thin service that:
- Receives approval requests from agents via a function call
- Persists the request with a correlation ID and timeout
- Posts a Slack message with approve/reject buttons
- Waits for the callback or timeout
- Returns the decision to the calling agent
This separation keeps your agents clean — they just call await request_approval(context) — and lets you swap Slack for Teams, email, or a custom UI later.
Setting up the Slack app
Create a Slack app at api.slack.com/apps with these scopes:
{
"bot_token_scopes": [
"chat:write",
"commands",
"channels:read",
"groups:read",
"im:read",
"mpim:read"
]
}Enable Interactivity & Shortcuts and set the request URL to https://your-domain.com/slack/interactions. Install the app to your workspace and note the Bot User OAuth Token (xoxb-...) and Signing Secret.
Store credentials in your secret manager, not .env. The signing secret verifies incoming callbacks; without it, anyone can forge approvals.
Building the approval manager
The manager needs three endpoints: one for agents to submit requests, one for Slack callbacks, and one for health checks. Here’s a FastAPI implementation:
# approval_manager/main.py
from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
from pydantic import BaseModel, Field
from typing import Optional, Literal
import httpx
import os
import uuid
import time
import json
from dataclasses import dataclass, asdict
app = FastAPI(title="AutoGen Approval Manager")
SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"]
SLACK_SIGNING_SECRET = os.environ["SLACK_SIGNING_SECRET"]
APPROVAL_CHANNEL = os.environ.get("APPROVAL_CHANNEL", "#approvals")
DEFAULT_TIMEOUT_SECONDS = int(os.environ.get("DEFAULT_TIMEOUT_SECONDS", "300"))
# In production, use Redis with TTL. This dict is for illustration.
PENDING_APPROVALS: dict[str, "ApprovalRequest"] = {}
class ApprovalRequest(BaseModel):
correlation_id: str
agent_name: str
title: str
options: list[str] = Field(default_factory=lambda: ["Approve", "Reject"])
timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS
metadata: dict = Field(default_factory=dict)
created_at: float = Field(default_factory=time.time)
slack_ts: Optional[str] = None
decision: Optional[str] = None
decided_at: Optional[float] = None
class AgentApprovalRequest(BaseModel):
agent_name: str
title: str
options: list[str] = ["Approve", "Reject"]
timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS
metadata: dict = {}
@app.post("/approvals/request")
async def request_approval(req: AgentApprovalRequest, background: BackgroundTasks):
correlation_id = str(uuid.uuid4())
approval = ApprovalRequest(
correlation_id=correlation_id,
agent_name=req.agent_name,
title=req.title,
description=req.description,
options=req.options,
timeout_seconds=req.timeout_seconds,
metadata=req.metadata,
)
PENDING_APPROVALS[correlation_id] = approval
# Post to Slack asynchronously
background.add_task(post_to_slack, approval)
# Wait for decision or timeout
return await wait_for_decision(correlation_id)
async def post_to_slack(approval: ApprovalRequest):
blocks = build_slack_blocks(approval)
async with httpx.AsyncClient() as client:
resp = await client.post(
"https://slack.com/api/chat.postMessage",
headers={"Authorization": f"Bearer {SLACK_BOT_TOKEN}"},
json={
"channel": APPROVAL_CHANNEL,
"blocks": blocks,
"text": f"Approval requested: {approval.title}",
},
)
data = resp.json()
if not data.get("ok"):
raise RuntimeError(f"Slack post failed: {data}")
approval.slack_ts = data["ts"]
def build_slack_blocks(approval: ApprovalRequest) -> list[dict]:
option_elements = [
{
"type": "button",
"text": {"type": "plain_text", "text": opt, "emoji": True},
"value": json.dumps({"correlation_id": approval.correlation_id, "decision": opt}),
"action_id": f"approval_{opt.lower()}",
"style": "primary" if opt.lower() == "approve" else "danger",
}
for opt in approval.options
]
return [
{
"type": "header",
"text": {"type": "plain_text", "text": f"🤖 {approval.title}", "emoji": True},
},
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": f"*Agent:*\n{approval.agent_name}"},
{"type": "mrkdwn", "text": f"*Timeout:*\n{approval.timeout_seconds}s"},
],
},
{"type": "section", "text": {"type": "mrkdwn", "text": approval.description}},
{"type": "divider"},
{"type": "actions", "elements": option_elements},
{
"type": "context",
"elements": [
{"type": "mrkdwn", "text": f"Correlation ID: `{approval.correlation_id}`"}
],
},
]
async def wait_for_decision(correlation_id: str) -> dict:
approval = PENDING_APPROVALS[correlation_id]
deadline = approval.created_at + approval.timeout_seconds
while time.time() < deadline:
if approval.decision:
return {
"correlation_id": correlation_id,
"decision": approval.decision,
"decided_at": approval.decided_at,
}
await asyncio.sleep(0.5)
# Timeout — update Slack message
await update_slack_timeout(approval)
approval.decision = "TIMEOUT"
approval.decided_at = time.time()
return {
"correlation_id": correlation_id,
"decision": "TIMEOUT",
"decided_at": approval.decided_at,
}
async def update_slack_timeout(approval: ApprovalRequest):
if not approval.slack_ts:
return
async with httpx.AsyncClient() as client:
await client.post(
"https://slack.com/api/chat.update",
headers={"Authorization": f"Bearer {SLACK_BOT_TOKEN}"},
json={
"channel": APPROVAL_CHANNEL,
"ts": approval.slack_ts,
"blocks": build_slack_blocks(approval) + [
{"type": "section", "text": {"type": "mrkdwn", "text": "⏰ *Timed out — no decision received*"}}
],
},
)
@app.post("/slack/interactions")
async def slack_interactions(request: Request):
# Verify signature in production!
form = await request.form()
payload = json.loads(form["payload"])
if payload["type"] != "block_actions":
return {"status": "ok"}
action = payload["actions"][0]
value = json.loads(action["value"])
correlation_id = value["correlation_id"]
decision = value["decision"]
approval = PENDING_APPROVALS.get(correlation_id)
if not approval:
return {"status": "ok"} # Already processed or expired
approval.decision = decision
approval.decided_at = time.time()
# Acknowledge to Slack immediately
await update_slack_decision(approval, decision, payload["user"]["username"])
return {"status": "ok"}
async def update_slack_decision(approval: ApprovalRequest, decision: str, username: str):
if not approval.slack_ts:
return
async with httpx.AsyncClient() as client:
await client.post(
"https://slack.com/api/chat.update",
headers={"Authorization": f"Bearer {SLACK_BOT_TOKEN}"},
json={
"channel": APPROVAL_CHANNEL,
"ts": approval.slack_ts,
"blocks": build_slack_blocks(approval) + [
{"type": "section", "text": {"type": "mrkdwn", "text": f"✅ *Decided by @{username}:* {decision}"}}
],
},
)
@app.get("/health")
async def health():
return {"status": "healthy", "pending": len(PENDING_APPROVALS)}Key points in this implementation:
- Correlation IDs link every callback to the original request. Generate them server-side; never trust client-supplied IDs.
- Background tasks post to Slack without blocking the agent’s HTTP response. The agent polls or uses a webhook for the result.
- Timeout handling updates the Slack message so approvers see the expired state. Without this, stale buttons linger and cause confusion.
- Signature verification is omitted for brevity. In production, validate
X-Slack-Signatureon every/slack/interactionscall using the signing secret.
Integrating with AutoGen agents
AutoGen agents call the approval manager through a function tool. Here’s a reusable pattern:
# autogen_integration/approval_tool.py
import httpx
import asyncio
from typing import Annotated
from autogen import AssistantAgent, UserProxyAgent
from autogen.tools import tool
APPROVAL_MANAGER_URL = "http://approval-manager:8000" # Service DNS name
@tool(
name="request_human_approval",
description="Request human approval for a decision. Returns the decision or TIMEOUT.",
)
async def request_human_approval(
title: Annotated[str, "Short title for the approval request"],
options: Annotated[list[str], "Available choices"] = ["Approve", "Reject"],
timeout_seconds: Annotated[int, "Seconds to wait"] = 300,
metadata: Annotated[dict, "Additional context for audit logs"] = {},
) -> dict:
async with httpx.AsyncClient(timeout=timeout_seconds + 10) as client:
# Submit request
resp = await client.post(
f"{APPROVAL_MANAGER_URL}/approvals/request",
json={
"agent_name": "current_agent", # Will be overridden by caller
"title": title,
"description": description,
"options": options,
"timeout_seconds": timeout_seconds,
"metadata": metadata,
},
)
resp.raise_for_status()
return resp.json()
# Usage in an agent
coding_agent = AssistantAgent(
name="coding_agent",
system_message="You are a senior engineer. Request approval before merging PRs.",
tools=[request_human_approval],
)
# The agent invokes it like:
# result = await request_human_approval(
# title="Merge PR #247",
# description="Refactors auth module. All tests pass. Breaking change: API v1 deprecated.",
# options=["Approve", "Request Changes", "Reject"],
# timeout_seconds=600,
# metadata={"pr_number": 247, "repo": "core-api"},
# )
# if result["decision"] == "Approve":
# # proceed with mergeThe agent passes its name via the agent_name field. In practice, inject this from the agent’s identity so audit logs show who asked.
Handling timeouts and retries
Timeouts are the most common failure mode. Design for three scenarios:
1. Approver never sees the message — Slack delivery failed or channel is muted.
Fix: Add a fallback notification (email, PagerDuty) after 60 seconds of no interaction. Track slack_ts presence; if null after the background task completes, the post failed.
2. Approver sees it but doesn’t act — Busy, away, or unclear request.
Fix: Escalate. After 50% of timeout, post a reminder in the thread. At 80%, @mention a backup approver from an on-call rotation.
3. Network partition between agent and manager — The agent’s HTTP request times out before the manager responds.
Fix: Make the agent’s call idempotent. Include a client-generated idempotency_key in the request. The manager returns the existing decision if the key matches a recent request.
# Idempotent request pattern
class AgentApprovalRequest(BaseModel):
idempotency_key: str # Client-generated UUID
# ... other fields ...
@app.post("/approvals/request")
async def request_approval(req: AgentApprovalRequest, background: BackgroundTasks):
# Check for existing request
existing = IDEMPOTENCY_STORE.get(req.idempotency_key)
if existing:
return await wait_for_decision(existing.correlation_id)
# ... create new ...
IDEMPOTENCY_STORE[req.idempotency_key] = approvalStore idempotency keys in Redis with a TTL matching your longest timeout plus buffer.
Common pitfalls
Pitfall: Blocking the agent’s event loop
AutoGen agents run in an async loop. A synchronous requests.post blocks everything. Use httpx.AsyncClient or aiohttp everywhere. The tool decorator handles async functions correctly.
Pitfall: Losing decisions on manager restart
The in-memory PENDING_APPROVALS dict evaporates on deploy. Use Redis with persistence enabled, or a database. Include a reconciliation job that scans for stale slack_ts entries and marks them MANAGER_RESTARTED so agents don’t hang forever.
Pitfall: Slack rate limits on high-volume workflows
chat.postMessage has tiered limits (typically 1 msg/sec per channel). Burst approvals will 429. Implement a token bucket per channel or use a queue (Celery, RQ) with retry/backoff. For n4n.ai users, the gateway’s built-in rate-limit handling can absorb provider-side 429s, but Slack’s limits are separate — handle them at the application layer.
Pitfall: Unverified callbacks
Without signature verification, a malicious actor can POST to /slack/interactions with forged decisions. Always verify:
from slack_sdk.signature import SignatureVerifier
verifier = SignatureVerifier(SLACK_SIGNING_SECRET)
@app.post("/slack/interactions")
async def slack_interactions(request: Request):
if not verifier.is_valid_request(
(await request.body()).decode(),
dict(request.headers),
):
raise HTTPException(401, "Invalid Slack signature")
# ... process ...Pitfall: Approval fatigue
If every minor decision routes to Slack, approvers ignore them. Reserve human-in-the-loop for irreversible or high-risk actions (production deploys, data deletion, spending > threshold). Use automated policy checks for the rest.
Tradeoffs and alternatives
| Approach | Latency | Audit Trail | Mobile | Complexity |
|---|---|---|---|---|
| Slack buttons | ~1s | Excellent | Native | Medium |
| Email + link | ~minutes | Good | Native | Low |
| Custom web UI | ~500ms | Full control | Responsive | High |
| Console input | ~0ms | None | No | Trivial |
Slack wins for team-facing workflows. Email works for external stakeholders without Slack access. A custom UI makes sense when you need complex forms (multi-field, file uploads, conditional logic) — but you’ll build authentication, session management, and mobile layouts yourself.
For multi-region deployments, deploy the approval manager in each region and route agents to the local instance. Slack is global; your manager shouldn’t be a single-point-of-failure.
Observability you’ll wish you had
Add these from day one:
- Structured logs with
correlation_id,agent_name,decision,latency_ms. Queryable in your log aggregator. - Metrics:
approval_requested_total,approval_decided_total{decision="approve|reject|timeout"},approval_latency_seconds. - Traces: Propagate a trace ID from the agent through the manager to Slack and back. OpenTelemetry auto-instrumentation covers FastAPI and httpx; add the correlation ID as a span attribute.
- Dead letter queue: Persist every callback payload to blob storage (S3, GCS) before processing. If a bug loses a decision, you can replay.
Testing the integration
Unit test the manager’s decision logic with a fake Slack client. Integration test the full flow against a real Slack workspace in CI — create a dedicated test channel and bot user. Use pytest-asyncio and a test container for Redis.
# tests/test_approval_flow.py
import pytest
from httpx import ASGITransport, AsyncClient
from approval_manager.main import app
@pytest.mark.asyncio
async def test_approval_timeout():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
resp = await client.post(
"/approvals/request",
json={
"agent_name": "test_agent",
"title": "Test",
"description": "Test",
"timeout_seconds": 1,
},
)
assert resp.status_code == 200
data = resp.json()
assert data["decision"] == "TIMEOUT"Run the integration test nightly, not on every PR — it needs network and credentials.
Wrapping up
You now have a production-grade pattern: agents call a function tool, the approval manager persists state and posts to Slack, humans click buttons, and the manager returns decisions with full auditability. The code above handles timeouts, idempotency, and Slack message updates — the pieces that cause incidents when omitted.
Start with the single-file manager, add Redis for persistence, then layer on observability and escalation. The autogen slack human approval integration becomes a reliable building block for any workflow that needs a human gate.