AgentFlow — AI Workflow Automation Agent

Production-style workflow automation backend built with FastAPI, OpenAI, LangChain, Redis, and a clean tool abstraction layer for CRM, calendar, and email integrations.

The agent receives a business trigger (like an inbound email), reasons about what actions to take, and autonomously executes a multi-step workflow — creating CRM leads, scheduling meetings, and sending confirmations — all without manual intervention.

The problem it solves

Every sales team has the same bottleneck: a potential client emails in, and someone has to manually read it, create a lead in the CRM, find an open calendar slot, schedule the meeting, and send a confirmation. That's 15 minutes of repetitive work per lead — multiplied across every inbound inquiry.

AgentFlow handles the entire sequence automatically. The LLM reads the email, decides what information to extract, and calls each integration in the correct order. Edge cases — ambiguous intent, missing information, API failures — are handled by the same reasoning engine, not by brittle if/else branches you have to maintain forever.

Why This Project

This project demonstrates practical AI agent engineering patterns that map directly to real automation problems:

• LLM function calling loop for dynamic, multi-step decision making
• Tool abstraction layer for swappable API integrations
• Real-time execution streaming via WebSocket
• Async workflow execution with background task management
• State persistence with Redis for run tracking
• Testable architecture with mocked LLM responses

Architecture

flowchart TD
    A[📧 Inbound Email\nPOST /trigger] --> B[API receives the event\nand creates a run ID]
    B --> C[Agent starts\nloads context + tools]
    C --> D[🧠 GPT-4o decides\nwhat to do next]
    D --> E{Did the agent\nfinish or need a tool?}

    E -- Needs a tool --> F[Tool Router\npicks the right integration]

    F --> G[📨 Read & analyze email\nparse_email_intent]
    F --> H[👤 Create lead in CRM\nHubSpot]
    F --> I[📅 Find open time slots\nGoogle Calendar]
    F --> J[📌 Book the meeting\nZoom + Google Calendar]
    F --> K[✉️ Send confirmation\nSendGrid]

    G & H & I & J & K --> L[Tool result sent back\nto the agent's memory]
    L --> D

    E -- Done, no more tools --> M[✅ Workflow complete\nall steps logged]

    C --> N[🔴 Live log stream\nWebSocket — you watch it happen]
    C --> O[💾 Run state saved\nRedis — survives restarts]

Loading

Tech Stack

• Backend: FastAPI, Uvicorn, Pydantic V2
• LLM: OpenAI GPT-4o with native function calling
• Agent framework: LangChain agents + custom execution loop
• State & messaging: Redis (key-value run state + Pub/Sub)
• Real-time: WebSocket streaming
• Testing: pytest, pytest-asyncio, httpx, unittest.mock

Project Structure

backend/
  main.py               # FastAPI app, routes, WebSocket, lifespan
  agent.py              # LLM function calling loop + event streaming
  tools.py              # Tool definitions + simulated API handlers
  models.py             # Pydantic schemas
  tests/
    conftest.py         # Shared async client fixture
    test_api.py         # HTTP endpoint tests
    test_tools.py       # Tool executor tests
    test_agent.py       # Agent loop tests with mocked OpenAI
  requirements.txt

frontend/
  index.html            # Interactive demo — no API key required

docker-compose.yml      # Redis + API in one command

Quick Start

1. Clone and install

git clone https://github.com/DavidFSantillan/AgentFlow.git
cd AgentFlow/backend

python -m venv .venv
# Windows PowerShell
.\.venv\Scripts\Activate.ps1
# macOS/Linux
# source .venv/bin/activate

pip install -r requirements.txt

2. Configure environment variables

Create a .env file inside backend/:

OPENAI_API_KEY=sk-...
REDIS_URL=redis://default:yourpassword@your-redis-host:6379

3. Run the API

cd backend
uvicorn main:app --reload --port 8000

• Swagger UI: http://localhost:8000/docs
• Health check: http://localhost:8000/health

Open frontend/index.html in your browser for the interactive demo — runs entirely without an API key.

API Example

Request

curl -X POST http://localhost:8000/trigger \
  -H "Content-Type: application/json" \
  -d '{
    "workflow_id": "email_to_crm_meeting",
    "data": {
      "from": "john.smith@acmecorp.com",
      "subject": "Interested in your platform",
      "body": "Hi, I saw your product and would love to schedule a demo this week."
    }
  }'

Response

{
  "run_id": "a3f9c112-84be-4e2a-b0cd-2f3a9e1d7c88",
  "status": "started",
  "message": "Agent is processing your request."
}

Then connect to the WebSocket to stream execution in real time:

ws://localhost:8000/ws/a3f9c112-84be-4e2a-b0cd-2f3a9e1d7c88

Each event looks like:

{ "type": "tool_result", "tool": "create_crm_lead", "result": { "lead_id": "LD-A9F3K2BX", "status": "created" } }
{ "type": "tool_result", "tool": "schedule_meeting", "result": { "meeting_id": "MTG-8823901122", "status": "scheduled" } }
{ "type": "complete", "steps_taken": 5 }

Testing

cd backend
pytest -v

With coverage:

pytest -v --cov=. --cov-report=html

The test suite covers all three layers independently — no real API key or Redis instance required:

File	What it tests
test_api.py	HTTP endpoints, status codes, validation
test_tools.py	Each tool executor returns the correct structure
test_agent.py	Agent loop logic with fully mocked OpenAI responses

Current Limitations

The agent currently runs in fire-and-forget mode — once triggered, it executes all steps autonomously without asking for confirmation. This is intentional for fully trusted, well-defined workflows, but it means:

• A misread email could create a duplicate CRM lead
• A meeting could be scheduled without the sales rep reviewing the slot
• There is no correction window between steps

The right solution for sensitive workflows is a Human-in-the-Loop (HITL) pattern, where the agent pauses after analysis and presents its proposed actions for approval before executing. This is the next planned addition.

Roadmap

• Human-in-the-Loop approval: Agent pauses before executing and sends a Slack message or dashboard notification with proposed actions — sales rep approves or corrects before anything is written to the CRM or calendar
• JWT authentication: Middleware for /trigger and /ws endpoints so only authorized services can trigger runs
• Real API integrations: Replace simulated tool handlers in tools.py with production clients — HubSpot SDK, Google Calendar API, SendGrid, Zoom
• Rate limiting: Redis-based rate limiting per API key to prevent abuse in multi-tenant deployments
• Observability: Emit traces to LangSmith or Langfuse for full agent step visibility and debugging
• Retry logic: Exponential backoff on tool calls for flaky external APIs