# Part 1: Building Agents from Scratch in Python 3

*Part of the* [*Multi Agent Orchestration 101 Series*](https://blog.htunnthuthu.com/ai-and-machine-learning/artificial-intelligence/multi-agent-orchestration-101)

## 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.

```python
# 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`.

```python
# 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.

```python
# 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:

```bash
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:

```python
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](https://blog.htunnthuthu.com/ai-and-machine-learning/artificial-intelligence/multi-agent-orchestration-101/part-2-tools-and-memory) — adding structured tool definitions, a tool dispatcher, and patterns for sharing context across agents.