From Solo Bot to Agent Fleet: A Practical Guide

#ai #devops #bots #python

There's a moment in every automation project where you realize your single bot isn't cutting it anymore. Maybe you started with a scraper that hits one API, or a simple Twitter engagement script. It worked fine for a while. But then the requirements grew—monitor ten sources, post to five platforms, verify a hundred accounts, handle payments.

Suddenly you're managing a fleet of agents, each with its own identity, API keys, and task queue. And you're still writing while True loops with time.sleep() hoping nothing crashes at 3 AM.

I've been there. This guide is about that transition—from a solo bot to a coordinated agent fleet. We'll cover the architecture, the tooling, and the real-world economics of running multiple agents at scale.

Why One Bot Isn't Enough

A single bot has a fundamental limitation: it's sequential. Even with async IO, you're bound by rate limits, IP reputation, and the simple fact that one logical unit can only be in one place at a time.

Consider a social media engagement bot. It needs to:

Like posts
Follow users
Reply to comments
Monitor DMs
Rotate proxies
Manage cooldowns

All of these compete for the same execution context. If your bot is busy replying, it's not monitoring. If it's monitoring, it's not rotating.

A fleet solves this by decoupling responsibilities. One agent handles discovery, another handles engagement, a third handles DM responses. Each runs independently, with its own rate limits and failure modes.

The Fleet Architecture

Here's the pattern I've settled on after a few painful rewrites:

┌─────────────┐     ┌──────────────┐
│  Orchestrator │────▶ Task Queue    │
└─────────────┘     └──────┬───────┘
                            │
          ┌─────────────────┼─────────────────┐
          ▼                 ▼                 ▼
    ┌──────────┐     ┌──────────┐     ┌──────────┐
    │ Agent A  │     │ Agent B  │     │ Agent C  │
    │ (scrape) │     │ (engage) │     │ (verify) │
    └──────────┘     └──────────┘     └──────────┘
          │                 │                 │
          └─────────────────┼─────────────────┘
                            ▼
                    ┌──────────────┐
                    │  Result Bus   │
                    └──────────────┘

The orchestrator doesn't do work—it delegates. It pushes tasks to a queue (Redis, RabbitMQ, or even a simple Postgres table). Agents pick up tasks, execute them, and push results back.

This is where the concept of agent-to-agent (A2A) delegation becomes real. One agent can spawn subtasks for another. Your scraper finds a new user → pushes a "verify this user" task to the verification agent → that agent completes it → pushes a "follow this user" task to the engagement agent.

Building a Simple Fleet Manager

Let's build a minimal fleet manager in Python. We'll use Redis for the task queue and asyncio for concurrent agent execution.

import asyncio
import json
import redis.asyncio as redis
from dataclasses import dataclass, asdict
from typing import Callable, Awaitable

@dataclass
class Task:
    agent_type: str
    action: str
    payload: dict
    priority: int = 0

class FleetManager:
    def __init__(self, redis_url: str = "redis://localhost"):
        self.redis = None
        self.agents: dict[str, Callable] = {}
        self.redis_url = redis_url

    async def connect(self):
        self.redis = redis.from_url(self.redis_url)

    def register_agent(self, agent_type: str, handler: Callable[[Task], Awaitable[dict]]):
        self.agents[agent_type] = handler

    async def push_task(self, task: Task):
        await self.redis.lpush(
            f"queue:{task.agent_type}",
            json.dumps(asdict(task))
        )

    async def worker_loop(self, agent_type: str):
        while True:
            _, data = await self.redis.brpop(f"queue:{agent_type}")
            task = Task(**json.loads(data))
            handler = self.agents.get(agent_type)
            if handler:
                try:
                    result = await handler(task)
                    await self.redis.publish(
                        f"results:{agent_type}",
                        json.dumps({"task_id": id(task), "result": result})
                    )
                except Exception as e:
                    print(f"Agent {agent_type} failed: {e}")
                    # Re-queue with backoff
                    await asyncio.sleep(5)
                    await self.push_task(task)

    async def start(self):
        await self.connect()
        workers = [
            self.worker_loop(agent_type)
            for agent_type in self.agents
        ]
        await asyncio.gather(*workers)

And here's how you'd register agents:

async def scraper_handler(task: Task):
    print(f"Scraping {task.payload['url']}")
    # ... scraping logic
    return {"status": "ok", "data": [...]}

async def engagement_handler(task: Task):
    print(f"Engaging with {task.payload['target']}")
    # ... engagement logic
    return {"status": "ok", "action": "liked"}

manager = FleetManager()
manager.register_agent("scraper", scraper_handler)
manager.register_agent("engagement", engagement_handler)

# Delegate from one agent to another
async def coordinated_scrape(task: Task):
    data = await scraper_handler(task)
    for item in data["data"]:
        await manager.push_task(Task(
            agent_type="engagement",
            action="like",
            payload={"target": item["id"]}
        ))
    return {"status": "ok", "dispatched": len(data["data"])}

Real Economics: Running a Fleet

Running agents costs money. Each agent needs a proxy, possibly a separate IP, and compute time. If you're running 100 agents, that's 100 proxies, 100 browser profiles, and the infrastructure to manage them.

This is where the economics get interesting. You can offset costs by having your agents earn while they work. For example, agents can complete microtasks—social signals, content verification, research—in between their primary duties.

A platform designed for this exact use case is roborent.cc. It's a marketplace where AI agents (and humans) earn USDT by completing tasks. The key features for fleet operators:

Multi-chain payouts: TRC-20, BEP-20, Arbitrum, TON—so you're not stuck with one network's fees
Fleet management: Dashboard for bot operators managing hundreds of agents
A2A delegation: Your agents can hire other agents to complete subtasks
Pro subscription: Removes fees and bumps your rate to 50 tasks/hour

The model is simple: your idle agents pick up paid tasks, earn USDT, and that income subsidizes your proxy costs. It's not passive income—it's active cost recovery.

Handling A2A Delegation at Scale

When one agent delegates to another, you need to handle:

Task serialization: The payload must be JSON-serializable
Idempotency: The same task should not be executed twice
Timeouts: Agents can hang; set TTL on tasks
Result routing: The delegating agent needs to know when its subtask is done

Here's a pattern for idempotent task execution:

import hashlib
import time

class IdempotentTask(Task):
    @property
    def id(self):
        raw = f"{self.agent_type}:{self.action}:{json.dumps(self.payload, sort_keys=True)}"
        return hashlib.sha256(raw.encode()).hexdigest()

async def execute_with_dedup(redis, task: IdempotentTask, handler):
    dedup_key = f"dedup:{task.id}"
    already_done = await redis.get(dedup_key)
    if already_done:
        return json.loads(already_done)

    result = await handler(task)
    await redis.setex(dedup_key, 3600, json.dumps(result))  # 1 hour TTL
    return result

Monitoring Your Fleet

You can't manage what you can't see. Every agent should emit structured logs and metrics. Here's a minimal setup:


python
class MonitoredAgent:
    def __init__(self, agent_type, redis):
        self.agent_type = agent_type
        self.redis = redis

    async def report(self, metric: str, value: float):
        await self.redis.hincrbyfloat(
            f"metrics:{self.agent_type}",
            metric,
            value
        )
        await self.redis.expire(f"metrics:{self.agent_type}", 86400)

    async def log(self