Part 2: Giving Agents Tools and Memory

Part of the Multi Agent Orchestration 101 Series

The Tool Problem I Kept Hitting

After getting my first agent loop working, I ran into a pattern that kept repeating: the agent would say it wanted to do something, but there was no clean way to map that decision to actual Python code. I was writing if "search" in response conditions everywhere. It was brittle and hard to extend.

The fix is structured tool definitions. This is the same mechanism that OpenAI's function calling and Anthropic's tool use are built on at the API level. Understanding it from first principles means you will know exactly what those APIs are doing when you start using them in Parts 3 and 4.

---

What Is a Tool?

A tool is a named, callable unit of work that an agent can invoke. It has three parts:

1. A schema — what the tool is called, what it does, what parameters it accepts

2. An implementation — the actual Python function

3. A dispatcher — code that maps a name + arguments to a function call

This separation is important. The schema is what you give to the LLM (or use in your rule-based _think() logic). The implementation and dispatcher are pure Python that the LLM never sees.

---

Defining Tools with JSON Schema

I use Python dataclass + dict for tool definitions. It is verbose but explicit — you can see exactly what will be sent to an LLM later.

# tools.py
from __future__ import annotations
import asyncio
import json
from dataclasses import dataclass
from typing import Any, Awaitable, Callable


@dataclass
class ToolDefinition:
    name: str
    description: str
    parameters: dict[str, Any]  # JSON Schema object
    fn: Callable[..., Awaitable[str]]

    def to_dict(self) -> dict[str, Any]:
        """Schema representation sent to OpenAI / Anthropic."""
        return {
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters,
        }

Here are two concrete tools I use in my own projects:

# --- Tool 1: run a shell command (carefully!) ---

async def run_shell(command: str) -> str:
    proc = await asyncio.create_subprocess_shell(
        command,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    if proc.returncode != 0:
        return f"Error (exit {proc.returncode}): {stderr.decode().strip()}"
    return stdout.decode().strip()


shell_tool = ToolDefinition(
    name="run_shell",
    description="Execute a shell command and return stdout. Use carefully.",
    parameters={
        "type": "object",
        "properties": {
            "command": {
                "type": "string",
                "description": "The shell command to execute",
            }
        },
        "required": ["command"],
    },
    fn=lambda args: run_shell(json.loads(args)["command"]),
)


# --- Tool 2: simple in-memory key/value store ---

_store: dict[str, str] = {}


async def kv_set(key: str, value: str) -> str:
    _store[key] = value
    return f"Stored '{key}'."


async def kv_get(key: str) -> str:
    if key not in _store:
        return f"Key '{key}' not found."
    return _store[key]


kv_set_tool = ToolDefinition(
    name="kv_set",
    description="Store a value under a key for later retrieval.",
    parameters={
        "type": "object",
        "properties": {
            "key": {"type": "string"},
            "value": {"type": "string"},
        },
        "required": ["key", "value"],
    },
    fn=lambda args: kv_set(**json.loads(args)),
)

kv_get_tool = ToolDefinition(
    name="kv_get",
    description="Retrieve a previously stored value by key.",
    parameters={
        "type": "object",
        "properties": {
            "key": {"type": "string"},
        },
        "required": ["key"],
    },
    fn=lambda args: kv_get(**json.loads(args)),
)

---

The Tool Dispatcher

The dispatcher maps a tool name to its ToolDefinition and calls it safely:

# dispatcher.py
from __future__ import annotations
from tools import ToolDefinition


class ToolDispatcher:
    def __init__(self, tools: list[ToolDefinition]) -> None:
        self._registry: dict[str, ToolDefinition] = {t.name: t for t in tools}

    def schemas(self) -> list[dict]:
        """Return all schemas — this is what you pass to OpenAI / Claude."""
        return [t.to_dict() for t in self._registry.values()]

    async def call(self, name: str, args: str) -> str:
        if name not in self._registry:
            return f"Unknown tool: '{name}'"
        try:
            return await self._registry[name].fn(args)
        except Exception as exc:
            return f"Tool error: {exc}"

Agents now hold a ToolDispatcher instead of a raw dict. The key benefit: dispatcher.schemas() returns exactly the JSON array that OpenAI and Anthropic expect for their tools parameter. No translation needed when you wire in the real API.

---

Short-Term vs Long-Term Memory

Part 1's memory list is short-term memory — it lives in RAM and resets when the process exits. That is fine for a single request, but falls apart when:

• An agent needs context from a previous session

• Two agents need to share state without passing messages

• You want to inspect what an agent was "thinking" after the fact

I use two patterns depending on the use case.

Pattern A: Shared Context Dict (in-process)

For agents in the same process that need a shared scratchpad:

# shared_context.py
from __future__ import annotations
from threading import Lock
from typing import Any


class SharedContext:
    """Thread-safe dictionary shared across all agents in one process."""

    def __init__(self) -> None:
        self._data: dict[str, Any] = {}
        self._lock = Lock()

    def set(self, key: str, value: Any) -> None:
        with self._lock:
            self._data[key] = value

    def get(self, key: str, default: Any = None) -> Any:
        with self._lock:
            return self._data.get(key, default)

    def snapshot(self) -> dict[str, Any]:
        with self._lock:
            return dict(self._data)

Pattern B: SQLite for Persistence

When I need history to survive restarts, I use SQLite via aiosqlite. It's a single file, zero infrastructure, and fast enough for local development and small production workloads.

# memory_store.py
from __future__ import annotations
import json
import time
import aiosqlite
from agent import Message

DB_PATH = "agent_memory.db"


async def init_db() -> None:
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute(
            """
            CREATE TABLE IF NOT EXISTS messages (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                agent_name TEXT NOT NULL,
                role TEXT NOT NULL,
                content TEXT NOT NULL,
                metadata TEXT DEFAULT '{}',
                created_at REAL NOT NULL
            )
            """
        )
        await db.commit()


async def save_message(agent_name: str, msg: Message) -> None:
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute(
            "INSERT INTO messages (agent_name, role, content, metadata, created_at) "
            "VALUES (?, ?, ?, ?, ?)",
            (agent_name, msg.role, msg.content, json.dumps(msg.metadata), time.time()),
        )
        await db.commit()


async def load_messages(agent_name: str, last_n: int = 20) -> list[Message]:
    async with aiosqlite.connect(DB_PATH) as db:
        async with db.execute(
            "SELECT role, content, metadata FROM messages "
            "WHERE agent_name = ? ORDER BY created_at DESC LIMIT ?",
            (agent_name, last_n),
        ) as cursor:
            rows = await cursor.fetchall()

    return [
        Message(role=r[0], content=r[1], metadata=json.loads(r[2]))
        for r in reversed(rows)
    ]

To use it, call load_messages at agent startup to restore context, and save_message after every new message.

---

Wiring It Together: An Agent with Tools and Persistence

Here is an updated Agent base class that incorporates the dispatcher and optional persistence:

# agent_v2.py
from __future__ import annotations
import asyncio
import json
from dataclasses import dataclass, field
from typing import Any

from dispatcher import ToolDispatcher
from memory_store import load_messages, save_message
from agent import Message


@dataclass
class AgentV2:
    name: str
    system_prompt: str
    dispatcher: ToolDispatcher
    persist: bool = False

    memory: list[Message] = field(default_factory=list)

    async def restore_memory(self) -> None:
        if self.persist:
            self.memory = await load_messages(self.name)

    async def run(self, user_message: str) -> str:
        msg = Message(role="user", content=user_message)
        self.memory.append(msg)
        if self.persist:
            await save_message(self.name, msg)

        for _ in range(10):
            raw = await self._think()

            if raw.startswith("DONE:"):
                final = raw[5:].strip()
                done_msg = Message(role="agent", content=final)
                self.memory.append(done_msg)
                if self.persist:
                    await save_message(self.name, done_msg)
                return final

            if raw.startswith("TOOL:"):
                _, name, args = raw.split(" ", 2)
                result = await self.dispatcher.call(name, args)
                tool_msg = Message(role="tool", content=result, metadata={"tool": name})
                self.memory.append(tool_msg)
                if self.persist:
                    await save_message(self.name, tool_msg)
                continue

            return raw

        return "Max iterations reached."

    async def _think(self) -> str:
        raise NotImplementedError

---

Sharing Context Between Two Agents

Here is a minimal complete example: two agents share a SharedContext. One writes a plan, the other executes it.

# two_agents_shared.py
import asyncio
from agent_v2 import AgentV2, Message
from dispatcher import ToolDispatcher
from shared_context import SharedContext
from tools import kv_set_tool, kv_get_tool

ctx = SharedContext()


class PlannerAgent(AgentV2):
    async def _think(self) -> str:
        # In a real system this is an LLM call
        last = self.memory[-1].content
        plan = f"Step 1: process '{last}'. Step 2: summarise."
        ctx.set("plan", plan)
        return f"DONE: plan stored"


class ExecutorAgent(AgentV2):
    async def _think(self) -> str:
        plan = ctx.get("plan")
        if not plan:
            return "DONE: no plan found"
        return f"DONE: executed — {plan}"


async def main() -> None:
    dispatcher = ToolDispatcher([kv_set_tool, kv_get_tool])

    planner = PlannerAgent(name="planner", system_prompt="", dispatcher=dispatcher)
    executor = ExecutorAgent(name="executor", system_prompt="", dispatcher=dispatcher)

    await planner.run("build feature X")
    result = await executor.run("go")
    print(result)
    # executed — Step 1: process 'build feature X'. Step 2: summarise.


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

---

Key Takeaways

• Separate the tool schema (what the LLM sees) from the implementation (Python)

• A ToolDispatcher with .schemas() makes LLM integration trivial later

• Short-term memory = list; long-term memory = SQLite (or Redis for scale)

• SharedContext is the simplest way to share state in a single process

---

Up Next

Part 3: OpenAI Multi-Agent Workflow — replacing the stub _think() with real OpenAI function calling, and building a supervisor that delegates to worker agents.