Building your first AI agent: a beginner's guide

Most tutorials show you a chat loop and call it an agent. That’s not an agent — that’s a wrapper. A real agent plans, acts, observes, and iterates without human intervention at each step. This guide walks through building a minimal but production-shaped agent from scratch, with runnable code at each layer.

What makes something an agent

An agent is a system that pursues a goal by selecting and executing actions in an environment, using feedback to adjust its plan. The minimal components:

1. Planner — decides what to do next given the goal and history
2. Tools — executable capabilities the planner can invoke
3. Memory — short-term (conversation) and long-term (facts, embeddings)
4. Executor — runs the loop, handles errors, enforces limits

The planner is usually an LLM with a structured prompt. Tools are typed functions with JSON schemas. Memory is whatever lets the agent not repeat itself or forget context.

Project structure

agent/
├── main.py              # entry point
├── planner.py           # LLM + prompt + parsing
├── tools/
│   ├── __init__.py
│   ├── registry.py      # tool definitions + dispatch
│   ├── web_search.py
│   ├── code_exec.py
│   └── file_ops.py
├── memory/
│   ├── __init__.py
│   ├── short_term.py    # conversation buffer
│   └── long_term.py     # vector store (optional)
└── executor.py          # run loop, retries, timeouts

Keep it flat. You’ll refactor later.

The tool registry

Tools are the agent’s hands. Define them as typed functions with JSON schemas the planner can reason about.

# tools/registry.py
from typing import Callable, Any
from dataclasses import dataclass
import json
import inspect

@dataclass
class Tool:
    name: str
    parameters: dict  # JSON Schema
    func: Callable

class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, Tool] = {}

    def register(self, name: str, description: str, parameters: dict):
        def decorator(func: Callable):
            self._tools[name] = Tool(name, description, parameters, func)
            return func
        return decorator

    def get_schemas(self) -> list[dict]:
        return [
            {
                "type": "function",
                "function": {
                    "name": t.name,
                    "description": t.description,
                    "parameters": t.parameters,
                },
            }
            for t in self._tools.values()
        ]

    def call(self, name: str, args: dict) -> Any:
        tool = self._tools.get(name)
        if not tool:
            raise ValueError(f"Unknown tool: {name}")
        return tool.func(**args)

registry = ToolRegistry()

Register tools with explicit schemas — don’t rely on automatic extraction. The planner needs to know exactly what each parameter means.

# tools/web_search.py
import requests
from tools.registry import registry

@registry.register(
    name="web_search",
    description="Search the web and return top results with snippets",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "Search query"},
            "max_results": {"type": "integer", "default": 5, "minimum": 1, "maximum": 10},
        },
        "required": ["query"],
    },
)
def web_search(query: str, max_results: int = 5) -> list[dict]:
    # Use your preferred search API (SerpAPI, Brave, etc.)
    resp = requests.post(
        "https://api.search.example/v1/search",
        json={"q": query, "count": max_results},
        timeout=10,
    )
    resp.raise_for_status()
    data = resp.json()
    return [
        {"title": r["title"], "url": r["url"], "snippet": r["snippet"]}
        for r in data.get("results", [])
    ]

The planner

The planner takes the goal, available tools, and conversation history, then returns either a tool call or a final answer. Use function calling if your provider supports it; otherwise parse structured output.

# planner.py
from typing import Literal
from dataclasses import dataclass
from tools.registry import registry

@dataclass
class PlanStep:
    type: Literal["tool_call", "final_answer"]
    tool_name: str | None = None
    tool_args: dict | None = None
    answer: str | None = None

SYSTEM_PROMPT = """You are an agent that solves tasks by calling tools.
You have access to these tools:
{tool_descriptions}

Think step by step. When you need information or action, call a tool.
When the task is complete, respond with a final answer.
Never make up tool results. If a tool fails, try a different approach.
"""

def build_planner_prompt(history: list[dict], goal: str) -> list[dict]:
    tool_descs = "\n".join(
        f"- {t['function']['name']}: {t['function']['description']}"
        for t in registry.get_schemas()
    )
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT.format(tool_descriptions=tool_descs)},
        *history,
        {"role": "user", "content": f"Goal: {goal}"},
    ]
    return messages

async def plan_next_step(
    client,  # OpenAI-compatible client
    model: str,
    history: list[dict],
    goal: str,
) -> PlanStep:
    messages = build_planner_prompt(history, goal)
    response = await client.chat.completions.create(
        model=model,
        messages=messages,
        tools=registry.get_schemas(),
        tool_choice="auto",
        temperature=0.1,
    )
    msg = response.choices[0].message
    if msg.tool_calls:
        tc = msg.tool_calls[0]
        return PlanStep(
            type="tool_call",
            tool_name=tc.function.name,
            tool_args=json.loads(tc.function.arguments),
        )
    return PlanStep(type="final_answer", answer=msg.content or "")

Keep temperature low. You want deterministic tool selection.

Memory: short-term and long-term

Short-term memory is the conversation buffer. Long-term memory lets the agent recall facts across sessions.

# memory/short_term.py
from collections import deque
from typing import Any

class ShortTermMemory:
    def __init__(self, max_turns: int = 20):
        self.max_turns = max_turns
        self.turns: deque[dict] = deque(maxlen=max_turns * 2)  # user + assistant pairs

    def add_user(self, content: str):
        self.turns.append({"role": "user", "content": content})

    def add_assistant(self, content: str, tool_calls: list | None = None):
        msg = {"role": "assistant", "content": content}
        if tool_calls:
            msg["tool_calls"] = tool_calls
        self.turns.append(msg)

    def add_tool_result(self, tool_call_id: str, name: str, result: Any):
        self.turns.append({
            "role": "tool",
            "tool_call_id": tool_call_id,
            "name": name,
            "content": json.dumps(result) if not isinstance(result, str) else result,
        })

    def get_history(self) -> list[dict]:
        return list(self.turns)

Long-term memory uses a vector store. Keep it simple — embed the goal + key facts, retrieve top-k on each planning step.

# memory/long_term.py
import numpy as np
from sentence_transformers import SentenceTransformer

class LongTermMemory:
    def __init__(self, embed_model: str = "all-MiniLM-L6-v2"):
        self.encoder = SentenceTransformer(embed_model)
        self.entries: list[dict] = []  # {"text": str, "embedding": np.ndarray, "metadata": dict}

    def add(self, text: str, metadata: dict | None = None):
        emb = self.encoder.encode(text, normalize_embeddings=True)
        self.entries.append({"text": text, "embedding": emb, "metadata": metadata or {}})

    def search(self, query: str, k: int = 3) -> list[dict]:
        if not self.entries:
            return []
        q_emb = self.encoder.encode(query, normalize_embeddings=True)
        scores = [float(q_emb @ e["embedding"]) for e in self.entries]
        top_idx = np.argsort(scores)[-k:][::-1]
        return [self.entries[i] for i in top_idx]

Inject retrieved memories into the planner prompt as context.

The executor loop

This is where the agent actually runs. Handle timeouts, retries, and step limits.

# executor.py
import asyncio
import json
import uuid
from dataclasses import dataclass
from typing import Any
from planner import plan_next_step, PlanStep
from memory.short_term import ShortTermMemory
from memory.long_term import LongTermMemory
from tools.registry import registry

@dataclass
class AgentResult:
    success: bool
    answer: str
    steps: int
    error: str | None = None

class AgentExecutor:
    def __init__(
        self,
        client,
        model: str,
        max_steps: int = 10,
        step_timeout: float = 30.0,
    ):
        self.client = client
        self.model = model
        self.max_steps = max_steps
        self.step_timeout = step_timeout
        self.short_term = ShortTermMemory()
        self.long_term = LongTermMemory()

    async def run(self, goal: str) -> AgentResult:
        # Seed long-term context if relevant
        relevant = self.long_term.search(goal)
        if relevant:
            ctx = "\n".join(f"[Memory] {r['text']}" for r in relevant)
            self.short_term.add_user(f"Relevant context:\n{ctx}")

        self.short_term.add_user(f"Goal: {goal}")

        for step_num in range(self.max_steps):
            try:
                step = await asyncio.wait_for(
                    plan_next_step(self.client, self.model, self.short_term.get_history(), goal),
                    timeout=self.step_timeout,
                )
            except asyncio.TimeoutError:
                return AgentResult(False, "", step_num, "Planner timeout")

            if step.type == "final_answer":
                self.long_term.add(f"Goal: {goal}\nAnswer: {step.answer}")
                return AgentResult(True, step.answer or "", step_num + 1)

            # Tool call
            tool_call_id = str(uuid.uuid4())
            self.short_term.add_assistant("", tool_calls=[{
                "id": tool_call_id,
                "type": "function",
                "function": {"name": step.tool_name, "arguments": json.dumps(step.tool_args)},
            }])

            try:
                result = await asyncio.wait_for(
                    asyncio.to_thread(registry.call, step.tool_name, step.tool_args),
                    timeout=self.step_timeout,
                )
                self.short_term.add_tool_result(tool_call_id, step.tool_name, result)
            except Exception as e:
                self.short_term.add_tool_result(tool_call_id, step.tool_name, f"Error: {e}")
                # Let the planner see the error and decide next step

        return AgentResult(False, "", self.max_steps, "Max steps exceeded")

Wiring it together

# main.py
import asyncio
import os
from openai import AsyncOpenAI
from executor import AgentExecutor

async def main():
    client = AsyncOpenAI(
        api_key=os.getenv("OPENROUTER_API_KEY"),
        base_url="https://openrouter.ai/api/v1",  # or your preferred gateway
    )
    agent = AgentExecutor(client=client, model="openai/gpt-4o-mini", max_steps=8)

    goal = "Find the current population of Tokyo and compare it to New York City"
    result = await agent.run(goal)

    if result.success:
        print(f"Answer: {result.answer}")
        print(f"Steps taken: {result.steps}")
    else:
        print(f"Failed: {result.error}")

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

Run it. Watch the logs. The agent will search for Tokyo population, search for NYC population, then synthesize the answer.

Common pitfalls

Tool schemas too vague. The planner hallucinates parameters. Be explicit: {"type": "integer", "minimum": 1, "maximum": 100} not just "type": "integer".

No error feedback to planner. If a tool fails, the planner must see the error message in the conversation. Otherwise it retries the same call indefinitely.

Infinite loops. The planner calls tool A, gets result, calls tool A again with same args. Add a step limit and consider tracking recent tool calls in the prompt.

Context overflow. Long conversations blow the context window. Summarize or truncate history. Keep the last N turns + a running summary.

Over-retrieval in long-term memory. Fetching 20 memories pollutes the prompt. Top-3 is usually enough. Rerank if you have the budget.

Sync tools in async loop. The executor uses asyncio.to_thread for blocking tools. If you have many concurrent agents, use a thread pool or rewrite tools as async.

Tradeoffs worth knowing

What to build next

1. Add a code execution tool with a sandbox (Docker, gVisor, or e2b). This turns the agent into a data analyst.
2. Persist long-term memory to disk (SQLite + FAISS, or a real vector DB).
3. Add observability — log every step, token count, latency, tool latency. You cannot debug what you cannot see.
4. Implement human-in-the-loop for high-stakes actions (send email, delete file, deploy).
5. Add a critic step — a second LLM call that reviews the plan before execution.

Closing note

The agent above is ~200 lines of Python. It has no framework lock-in, no magic, and every component is replaceable. That’s the point. Frameworks (LangGraph, AutoGen, CrewAI) solve coordination at scale. For your first agent, they add indirection without clarity. Build the loop yourself once. Then decide what you actually need from a framework.