The Complete Guide to AI Agent Architecture for Business Automation

Script: "If email contains 'invoice', save attachment to /invoices/"
Agent: "Read this email, understand the context, decide if it's an invoice, 
       extract the data, verify against our records, flag discrepancies, 
       and route to the right person."

                    ┌─────────────┐
                    │  Orchestrator │
                    │  (Decision    │
                    │   Engine)     │
                    └───────┬───────┘
                            │
            ┌───────────────┼───────────────┐
            │               │               │
     ┌──────┴──────┐ ┌─────┴──────┐ ┌─────┴──────┐
     │  Perception  │ │  Reasoning  │ │   Action    │
     │  Layer       │ │  Layer      │ │   Layer     │
     └──────────────┘ └────────────┘ └────────────┘
     - Read emails     - Analyze       - Send emails
     - Parse docs      - Decide        - Create tasks
     - Monitor feeds   - Plan          - Update CRM
     - Watch events    - Prioritize    - Generate docs

class PerceptionLayer:
    def __init__(self):
        self.sources = []

    def add_source(self, name, fetcher):
        self.sources.append({"name": name, "fetch": fetcher})

    def gather_context(self):
        context = {}
        for source in self.sources:
            try:
                context[source["name"]] = source["fetch"]()
            except Exception as e:
                context[source["name"]] = {"error": str(e)}
        return context

# Configure sources
perception = PerceptionLayer()
perception.add_source("emails", lambda: gmail.get_unread(max=20))
perception.add_source("calendar", lambda: gcal.get_today())
perception.add_source("crm", lambda: hubspot.get_recent_activity())
perception.add_source("slack", lambda: slack.get_mentions())

import anthropic

class ReasoningLayer:
    def __init__(self):
        self.client = anthropic.Anthropic()
        self.memory = []  # Short-term memory

    def analyze(self, context, goal):
        prompt = f"""You are a business operations agent.

Current context:
{json.dumps(context, indent=2)}

Recent actions taken:
{json.dumps(self.memory[-10:], indent=2)}

Goal: {goal}

Decide what actions to take. For each action, specify:
1. action_type: email_reply | create_task | update_crm | send_slack | generate_report | escalate
2. target: who/what
3. content: what to do
4. priority: 1-5
5. reasoning: why this action

Return as JSON array of actions, ordered by priority.
"""

        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2000,
            messages=[{"role": "user", "content": prompt}]
        )

        actions = json.loads(response.content[0].text)
        self.memory.extend(actions)
        return actions

class ActionLayer:
    def __init__(self):
        self.handlers = {}

    def register(self, action_type, handler):
        self.handlers[action_type] = handler

    def execute(self, actions):
        results = []
        for action in sorted(actions, key=lambda x: x["priority"]):
            handler = self.handlers.get(action["action_type"])
            if handler:
                try:
                    result = handler(action)
                    results.append({"action": action, "status": "success", "result": result})
                except Exception as e:
                    results.append({"action": action, "status": "error", "error": str(e)})
            else:
                results.append({"action": action, "status": "no_handler"})
        return results

# Register handlers
actions = ActionLayer()
actions.register("email_reply", lambda a: gmail.send(a["target"], a["content"]))
actions.register("create_task", lambda a: asana.create_task(a["content"]))
actions.register("update_crm", lambda a: hubspot.update(a["target"], a["content"]))
actions.register("send_slack", lambda a: slack.post(a["target"], a["content"]))
actions.register("escalate", lambda a: notify_human(a["target"], a["content"]))

class BusinessAgent:
    def __init__(self):
        self.perception = PerceptionLayer()
        self.reasoning = ReasoningLayer()
        self.action = ActionLayer()
        self.setup_sources()
        self.setup_handlers()

    def run_cycle(self, goal="Manage daily business operations efficiently"):
        # 1. Perceive
        context = self.perception.gather_context()

        # 2. Reason
        planned_actions = self.reasoning.analyze(context, goal)

        # 3. Act
        results = self.action.execute(planned_actions)

        # 4. Learn (update memory)
        self.reasoning.memory.append({
            "cycle": datetime.now().isoformat(),
            "actions_taken": len(results),
            "successes": sum(1 for r in results if r["status"] == "success")
        })

        return results

# Run every 15 minutes
agent = BusinessAgent()
schedule.every(15).minutes.do(agent.run_cycle)

HIGH_STAKES = ["send_proposal", "modify_pricing", "client_communication", "financial"]

def execute_with_approval(action):
    if action["action_type"] in HIGH_STAKES or action.get("value", 0) > 1000:
        # Queue for human review
        send_approval_request(action)
        return {"status": "pending_approval"}
    else:
        return action_layer.execute([action])