Part 1: Building Agents from Scratch in Python 3

Part of the Multi Agent Orchestration 101 Series

The Moment I Stopped Using Frameworks (Temporarily)

I had been using LangChain for a few months. It solved problems quickly, but every time something broke I had no idea where to look. The stack traces pointed at internal abstractions, not my code. I spent more time reading framework source than actually building.

So I opened a blank .py file and asked myself: what is the minimum amount of code needed for two programs to act as agents and coordinate?

Turns out — not much. This part shows you exactly that minimum. No frameworks. No magic. Just Python 3 and the standard library.

---

What is an Agent, Actually?

Before writing a single line, I want to be precise about vocabulary because "agent" means different things to different people.

For this series, an agent is a program that:

1. Perceives an input (a goal, a message, a tool result)

2. Decides what to do next (call a tool, ask another agent, respond)

3. Acts (executes the decision)

4. Loops until a stopping condition is met

That's it. An LLM is just the "decide" step. The rest is Python.

A multi-agent system is two or more agents that coordinate — sharing messages, delegating subtasks, or checking each other's work.

---

The Minimal Agent Loop

Let's build the simplest possible agent. No LLM yet — just the skeleton.

# agent.py
from __future__ import annotations
import asyncio
from dataclasses import dataclass, field
from typing import Any, Callable, Awaitable


@dataclass
class Message:
    role: str          # "user" | "agent" | "tool"
    content: str
    metadata: dict[str, Any] = field(default_factory=dict)


@dataclass
class Agent:
    name: str
    system_prompt: str
    tools: dict[str, Callable[..., Awaitable[str]]] = field(default_factory=dict)
    memory: list[Message] = field(default_factory=list)

    async def run(self, user_message: str) -> str:
        """Main agent loop: perceive → decide → act → loop."""
        self.memory.append(Message(role="user", content=user_message))

        for _ in range(10):  # max iterations to prevent infinite loops
            response = await self._think()

            if response.startswith("DONE:"):
                return response[5:].strip()

            if response.startswith("TOOL:"):
                tool_result = await self._use_tool(response)
                self.memory.append(Message(role="tool", content=tool_result))
                continue

            # Plain response — we're done
            return response

        return "Max iterations reached."

    async def _think(self) -> str:
        """Decide what to do next. Subclasses replace this with an LLM call."""
        raise NotImplementedError

    async def _use_tool(self, tool_call: str) -> str:
        """Parse 'TOOL: <name> <args>' and dispatch."""
        parts = tool_call[5:].strip().split(" ", 1)
        name = parts[0]
        args = parts[1] if len(parts) > 1 else ""

        if name not in self.tools:
            return f"Error: tool '{name}' not found."

        return await self.tools[name](args)

This is the core loop. _think() is intentionally abstract — Part 3 and 4 replace it with OpenAI and Claude calls respectively.

---

Adding a Message Bus

For two agents to coordinate, they need a way to pass messages. The simplest approach is an asyncio.Queue.

# message_bus.py
import asyncio
from agent import Message


class MessageBus:
    """Simple in-process message bus. Each agent subscribes to a named inbox."""

    def __init__(self) -> None:
        self._queues: dict[str, asyncio.Queue[Message]] = {}

    def register(self, agent_name: str) -> None:
        self._queues[agent_name] = asyncio.Queue()

    async def send(self, to: str, message: Message) -> None:
        if to not in self._queues:
            raise KeyError(f"No inbox for agent '{to}'")
        await self._queues[to].put(message)

    async def receive(self, agent_name: str, timeout: float = 5.0) -> Message | None:
        try:
            return await asyncio.wait_for(
                self._queues[agent_name].get(), timeout=timeout
            )
        except asyncio.TimeoutError:
            return None

This is a deliberately trivial bus. In production (Part 5) you would swap it for Redis Streams or a proper message broker. But for learning, the queue is perfect because you can see every message in memory with a debugger.

---

A Concrete Example: Two Echo Agents

Let me prove the plumbing works before adding LLM complexity. Here are two agents that pass a counter back and forth.

# echo_agents.py
import asyncio
from agent import Agent, Message
from message_bus import MessageBus


class EchoAgent(Agent):
    """Toy agent that receives a number and sends back number + 1."""

    def __init__(self, name: str, bus: MessageBus, partner: str) -> None:
        super().__init__(name=name, system_prompt="")
        self.bus = bus
        self.partner = partner

    async def _think(self) -> str:
        # Last message in memory is what we should respond to
        last = self.memory[-1]
        value = int(last.content)
        if value >= 5:
            return f"DONE: reached {value}"
        response = str(value + 1)
        await self.bus.send(
            self.partner,
            Message(role="agent", content=response, metadata={"from": self.name}),
        )
        return f"TOOL: wait"  # pause, let partner respond

    async def listen_loop(self) -> None:
        while True:
            msg = await self.bus.receive(self.name)
            if msg is None:
                break
            result = await self.run(msg.content)
            if result:
                print(f"[{self.name}] finished: {result}")
                break


async def main() -> None:
    bus = MessageBus()
    bus.register("alice")
    bus.register("bob")

    alice = EchoAgent("alice", bus, partner="bob")
    bob = EchoAgent("bob", bus, partner="alice")

    # Kick off the conversation
    await bus.send("alice", Message(role="user", content="0"))

    await asyncio.gather(
        alice.listen_loop(),
        bob.listen_loop(),
    )


if __name__ == "__main__":
    asyncio.run(main())

Run it:

python echo_agents.py
# [alice] finished: reached 5

Alice receives 0, sends 1 to Bob. Bob receives 1, sends 2 back. This continues until the counter hits 5. Two agents, coordinating, with explicit message passing. No framework.

---

Agent Memory: Short-Term and Context Windows

The memory list in Agent is short-term memory — it resets each time you create a new instance. This mirrors an LLM's context window. Things to know:

• Keep it bounded. An unbounded list will eventually overflow an LLM's context. I add a max_memory trim in my real projects.

• Role matters. The role field maps directly to LLM message roles: user, assistant, tool. Getting this right is what makes LLM integration clean.

• Don't store secrets. Memory is often serialised for debugging. Strip API keys, tokens, and PII before they enter an agent's context.

Here's the trimming pattern I use:

def _trim_memory(self, keep_last: int = 20) -> None:
    """Keep the first message (system context) and the last N messages."""
    if len(self.memory) > keep_last + 1:
        self.memory = [self.memory[0]] + self.memory[-(keep_last):]

---

What's Missing (On Purpose)

I deliberately left out:

• LLM calls — covered in Parts 3 and 4

• Persistent storage — covered in Part 2

• Error handling — covered in Part 5

• Parallel execution — covered in Part 3

The goal of this part is to have the skeleton clear in your head before adding complexity. If you can explain what _think(), memory, and the message bus do, you are ready for the next part.

---

Key Takeaways

• An agent is a loop: perceive → decide → act → repeat

• Multi-agent coordination is message passing — start with a queue

• The LLM is just the "decide" step; everything else is Python

• Build the skeleton first, add intelligence second

---

Up Next

Part 2: Tools and Memory — adding structured tool definitions, a tool dispatcher, and patterns for sharing context across agents.