Set up the project skeleton.

Before any code, we need three things: a corpus (the documents the agent will read), a place to put our agent code, and a Python environment with an SDK installed. That's it. No vector database, no framework, no orchestration layer. Every piece of infrastructure you add is one more thing hiding the primitives you're trying to learn.

Picking a corpus

A corpus is just a folder of text files. Pick something you care about — you'll know immediately when the agent is wrong. Good choices: the PostgreSQL documentation, your own markdown notes, or a single open-source project's docs (React, Rust book, Kubernetes concepts). Aim for 50–500 markdown files.

Fewer than 50 and the agent rarely needs to think; more than 500 and you'll spend too much time waiting on retrieval iterations. For this tutorial we'll assume PostgreSQL docs.

The folder layout

# Create the project
mkdir research-agent && cd research-agent
mkdir -p corpus agent evals scripts runs

# Files we'll fill in over the next steps
touch agent/__init__.py
touch agent/loop.py        # the agentic loop
touch agent/tools.py       # tool definitions + handlers
touch agent/prompts.py     # system prompts
touch scripts/run.py       # CLI entry point
touch .env                 # for API keys

Run tree (or ls -R) and you'll see:

research-agent/
├── agent/
│   ├── __init__.py
│   ├── loop.py
│   ├── prompts.py
│   └── tools.py
├── corpus/         # populate with your .md files
├── evals/
├── runs/
├── scripts/
│   └── run.py
└── .env

Installing dependencies

The Python environment is identical for both APIs — only the SDK package differs.

# Create a virtual environment
python -m venv .venv
source .venv/bin/activate

# Install Anthropic SDK + utilities
pip install anthropic python-dotenv rich

# Create a virtual environment
python -m venv .venv
source .venv/bin/activate

# Install OpenAI SDK + utilities
pip install openai python-dotenv rich

Three packages, only one provider-specific. python-dotenv reads your API key from .env. rich makes traces readable.

The .env file

# .env — get this from console.anthropic.com
ANTHROPIC_API_KEY=sk-ant-api03-...your-key...

# .env — get this from platform.openai.com/api-keys
OPENAI_API_KEY=sk-proj-...your-key...

Add .env to .gitignore immediately — keys checked into git get scraped within hours.

Define three tools. Exactly three.

An agent is a model that calls tools in a loop. Without tools, you have a chatbot. With tools, you have an agent. Three tools is the minimum that lets a research agent show its reasoning.

The three tools and why

search_docs(query) finds relevant documents — returns doc IDs plus snippets, not full text. Snippets keep context lean and force the agent to decide which docs are worth fetching.

fetch_doc(doc_id) reads a full document. Separating "find" from "read" is deliberate: it makes the agent's relevance judgments visible in the trace.

submit_answer(answer, citations) ends the loop with structured output. Without this, the agent might just say the answer in plain text and we'd have no way to extract citations.

Writing the tool schemas

Both APIs use JSON Schema for parameters, but the wrapping differs. Anthropic wraps in {name, description, input_schema}. OpenAI's Responses API uses {type: "function", name, description, parameters}. The schemas themselves are identical.

# agent/tools.py
TOOLS = [
    {
        "name": "search_docs",
        "description": (
            "Search the corpus by keyword. Returns up to "
            "5 matches, each with doc_id and 300-char snippet."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"}
            },
            "required": ["query"],
        },
    },
    {
        "name": "fetch_doc",
        "description": "Fetch full text of a doc by doc_id.",
        "input_schema": {
            "type": "object",
            "properties": {
                "doc_id": {"type": "string"}
            },
            "required": ["doc_id"],
        },
    },
    {
        "name": "submit_answer",
        "description": "Submit final answer. Ends conversation.",
        "input_schema": {
            "type": "object",
            "properties": {
                "answer": {"type": "string"},
                "citations": {
                    "type": "array",
                    "items": {"type": "string"},
                },
            },
            "required": ["answer", "citations"],
        },
    },
]

# agent/tools.py
TOOLS = [
    {
        "type": "function",
        "name": "search_docs",
        "description": (
            "Search the corpus by keyword. Returns up to "
            "5 matches, each with doc_id and 300-char snippet."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string"}
            },
            "required": ["query"],
            "additionalProperties": False,
        },
    },
    {
        "type": "function",
        "name": "fetch_doc",
        "description": "Fetch full text of a doc by doc_id.",
        "parameters": {
            "type": "object",
            "properties": {
                "doc_id": {"type": "string"}
            },
            "required": ["doc_id"],
            "additionalProperties": False,
        },
    },
    {
        "type": "function",
        "name": "submit_answer",
        "description": "Submit final answer. Ends conversation.",
        "parameters": {
            "type": "object",
            "properties": {
                "answer": {"type": "string"},
                "citations": {
                    "type": "array",
                    "items": {"type": "string"},
                },
            },
            "required": ["answer", "citations"],
            "additionalProperties": False,
        },
    },
]

Writing the handlers

Handlers are pure Python — no provider-specific code. They're the same for both APIs.

# agent/tools.py (continued — shared)
from pathlib import Path

CORPUS = Path("corpus")

def search_docs(query: str) -> list[dict]:
    """Dumb substring scan. Intentionally bad."""
    results = []
    q = query.lower()
    for path in CORPUS.glob("*.md"):
        text = path.read_text(encoding="utf-8")
        if q in text.lower():
            idx = text.lower().find(q)
            start = max(0, idx - 100)
            end = min(len(text), idx + 200)
            results.append({
                "doc_id": path.stem,
                "snippet": text[start:end].strip(),
            })
        if len(results) >= 5: break
    return results

def fetch_doc(doc_id: str) -> dict:
    path = CORPUS / f"{doc_id}.md"
    if not path.exists():
        return {"error": f"no doc: {doc_id}"}
    return {
        "doc_id": doc_id,
        "content": path.read_text(encoding="utf-8"),
    }

HANDLERS = {
    "search_docs": search_docs,
    "fetch_doc": fetch_doc,
    # submit_answer is handled in the loop, not here
}

Testing the tools in isolation

Before touching the loop, prove the tools work. Open a Python REPL:

>>> from agent.tools import search_docs
>>> results = search_docs("connection pool")
>>> for r in results:
...     print(r["doc_id"], "→", r["snippet"][:60])

runtime-config-connection → ...connection pool can hold up to max_connections...
runtime-config-resource → ...each connection consumes shared memory, so pool size...
pgbouncer-modes → ...connection pooling in PgBouncer operates in three distinct...
libpq-connect → ...PQconnectdb opens a single connection; for connection pool...
ddl-system-columns → ...system catalogs in the connection pool are shared...

Five real results, each with a snippet. Crude but functional.

Write the loop. Make it ugly.

The core idea of agents: a while-loop where each iteration calls the model, executes any tools it asked for, and feeds results back. Everything else — frameworks, orchestration, agent SDKs — is decoration around this loop.

The mental model

The history list grows on every iteration. The model sees the full history on every turn — it remembers what it searched, what came back, what it decided. That's the agent's "memory" for this conversation.

This mental model is identical for both APIs. They differ in names: Anthropic calls the history messages, OpenAI calls it input. Anthropic returns content blocks; OpenAI returns output items. Anthropic uses tool_use/tool_result; OpenAI uses function_call/function_call_output. Same shape, different vocabulary.

The system prompt

Identical for both APIs.

# agent/prompts.py
SYSTEM_PROMPT = """You are a research assistant for a documentation corpus.

You have three tools:
- search_docs(query): find relevant documents
- fetch_doc(doc_id): read a full document
- submit_answer(answer, citations): finish

Process:
1. Search for terms related to the user's question.
2. If a snippet looks promising, fetch the full doc.
3. Repeat until you have enough to answer confidently.
4. Submit your answer with citations to doc_ids you used.

Rules:
- Every claim must be supported by a cited doc.
- If the corpus doesn't have the answer, say so honestly.
- Don't search for the same thing twice.
- Aim for 3-6 tool calls before submitting."""

The loop itself

# agent/loop.py — Anthropic Messages API
from anthropic import Anthropic
from agent.tools import TOOLS, HANDLERS
from agent.prompts import SYSTEM_PROMPT

client = Anthropic()

def run_agent(user_query: str, max_steps: int = 10):
    messages = [{"role": "user", "content": user_query}]
    trace = []

    for step in range(max_steps):
        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=4096,
            system=SYSTEM_PROMPT,
            tools=TOOLS,
            messages=messages,
        )

        # Append assistant turn — required by API
        messages.append({
            "role": "assistant",
            "content": response.content,
        })
        trace.append({"step": step, "response": response})

        if response.stop_reason == "end_turn":
            return {"status": "halted_no_answer",
                    "trace": trace}

        tool_results = []
        for block in response.content:
            if block.type != "tool_use":
                continue

            if block.name == "submit_answer":
                return {
                    "status": "answered",
                    "answer": block.input["answer"],
                    "citations": block.input["citations"],
                    "steps_used": step + 1,
                    "trace": trace,
                }

            try:
                result = HANDLERS[block.name](**block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(result),
                })
            except Exception as e:
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": f"error: {e}",
                    "is_error": True,
                })

        # All tool results in one user turn
        messages.append({"role": "user",
                         "content": tool_results})

    return {"status": "step_limit", "trace": trace}

# agent/loop.py — OpenAI Responses API
import json
from openai import OpenAI
from agent.tools import TOOLS, HANDLERS
from agent.prompts import SYSTEM_PROMPT

client = OpenAI()

def run_agent(user_query: str, max_steps: int = 10):
    # Responses API uses a flat list of "items"
    input_items = [
        {"role": "user", "content": user_query}
    ]
    trace = []

    for step in range(max_steps):
        response = client.responses.create(
            model="gpt-5.5",
            instructions=SYSTEM_PROMPT,
            tools=TOOLS,
            input=input_items,
        )

        # Append every output item to history
        for item in response.output:
            input_items.append(item.model_dump())
        trace.append({"step": step, "response": response})

        # Find function_call items in this turn
        calls = [i for i in response.output
                 if i.type == "function_call"]

        if not calls:
            # Model produced text without calling a tool
            return {"status": "halted_no_answer",
                    "trace": trace}

        for call in calls:
            args = json.loads(call.arguments)

            if call.name == "submit_answer":
                return {
                    "status": "answered",
                    "answer": args["answer"],
                    "citations": args["citations"],
                    "steps_used": step + 1,
                    "trace": trace,
                }

            try:
                result = HANDLERS[call.name](**args)
                output = str(result)
            except Exception as e:
                output = f"error: {e}"

            # Match output to call by call_id
            input_items.append({
                "type": "function_call_output",
                "call_id": call.call_id,
                "output": output,
            })

    return {"status": "step_limit", "trace": trace}

Same logic — different shapes

The loops are structurally identical: send history + tools, append response, dispatch any tool calls, append results, repeat. But the data shapes differ in ways worth understanding.

Anthropic uses messages with content blocks. Each turn is a {role, content} object where content is either a string (for user input) or a list of blocks (for tool calls, results, and text). The API enforces strict turn alternation: assistant turns must come right before any user turn that carries tool_results.

OpenAI Responses uses a flat item list. The input parameter is a list of items, each with a type: user messages, function_call items, function_call_output items, and so on. No strict alternation; items are correlated by call_id instead of by position.

The most common bug, in both APIs

anthropic.BadRequestError: Error code: 400 -
{'error': {'type': 'invalid_request_error',
'message': 'messages.1: tool_result block found without
corresponding tool_use block'}}

openai.BadRequestError: Error code: 400 -
{'error': {'message': 'No tool call found for
function_call_output with call_id call_xyz...',
'type': 'invalid_request_error'}}

Different error messages, same underlying mistake: tool results need their corresponding calls in the history.

Run it. Watch every trace. Take notes.

Now we run the agent on real questions and read what it does, line by line. This is the most important step in Phase 1. Everything in Phases 2 and 3 is a response to behavior we see here.

Example 1: a clean run

$ python scripts/run.py "What's the default PostgreSQL port?"

Here's the trace you'd see — identical regardless of which API you used:

──────────────────── Step 0 ────────────────────
┌─ thinking ─────────────────────────────────────┐
│ I'll search for "default port" to find the     │
│ PostgreSQL configuration.                      │
└────────────────────────────────────────────────┘
→ search_docs({'query': 'default port'})

──────────────────── Step 1 ────────────────────
┌─ thinking ─────────────────────────────────────┐
│ The snippet from runtime-config-connection     │
│ mentions port 5432. Let me fetch the full doc  │
│ to confirm.                                    │
└────────────────────────────────────────────────┘
→ fetch_doc({'doc_id': 'runtime-config-connection'})

──────────────────── Step 2 ────────────────────
→ submit_answer({
    'answer': 'PostgreSQL listens on TCP port
    5432 by default. This can be changed via the
    `port` parameter in postgresql.conf.',
    'citations': ['runtime-config-connection']
  })

─────────────── Final answer ───────────────
PostgreSQL listens on TCP port 5432 by default.
This can be changed via the `port` parameter in
postgresql.conf.
citations: ['runtime-config-connection']
status: answered

Example 2: a multi-hop question

$ python scripts/run.py "How does PgBouncer's transaction pooling
   interact with prepared statements?"

──────────────────── Step 0 ────────────────────
→ search_docs({'query': 'PgBouncer transaction pooling'})

──────────────────── Step 1 ────────────────────
┌─ thinking ─────────────────────────────────────┐
│ Got pgbouncer-modes. I need to fetch the full  │
│ doc AND search for prepared statements.        │
└────────────────────────────────────────────────┘
→ fetch_doc({'doc_id': 'pgbouncer-modes'})

──────────────────── Step 2 ────────────────────
→ search_docs({'query': 'prepared statements'})

──────────────────── Step 3 ────────────────────
→ fetch_doc({'doc_id': 'sql-prepare'})

──────────────────── Step 4 ────────────────────
┌─ thinking ─────────────────────────────────────┐
│ Prepared statements are session-scoped. In     │
│ transaction pooling, server connections rotate │
│ between transactions. They conflict.           │
└────────────────────────────────────────────────┘
→ submit_answer({
    'answer': 'Transaction pooling and prepared
    statements conflict. Prepared statements are
    scoped to a session, but transaction pooling
    rotates server connections between
    transactions...',
    'citations': ['pgbouncer-modes', 'sql-prepare']
  })

status: answered (5 steps)

Example 3: a failure mode you'll definitely see

Now a question where the agent goes wrong. Crude substring search makes Phase 1 failure modes show up clearly:

$ python scripts/run.py "When should I VACUUM versus VACUUM FULL?"

──────────────────── Step 0 ────────────────────
→ search_docs({'query': 'VACUUM'})

──────────────────── Step 1 ────────────────────
┌─ thinking ─────────────────────────────────────┐
│ Got 5 results but they all mention VACUUM      │
│ briefly. Let me search more specifically.      │
└────────────────────────────────────────────────┘
→ search_docs({'query': 'VACUUM FULL'})

──────────────────── Step 2 ────────────────────
→ search_docs({'query': 'when to use VACUUM'})

──────────────────── Step 3 ────────────────────
┌─ thinking ─────────────────────────────────────┐
│ The third search returned nothing — substring  │
│ "when to use" doesn't appear literally. Let    │
│ me try fetching routine-vacuuming directly.    │
└────────────────────────────────────────────────┘
→ fetch_doc({'doc_id': 'routine-vacuuming'})

──────────────────── Step 4 ────────────────────
→ submit_answer({...})

status: answered (5 steps)

The trace log

Don't just watch traces — write them down. Keep runs/notes.md:

# Phase 1 trace log

## 2026-05-16 — first runs

### Q: "default postgres port"
- 3 steps, clean. ✓ search → fetch → submit

### Q: "PgBouncer pooling + prepared statements"
- 5 steps, correct answer
- ✓ decomposed into two sub-searches

### Q: "VACUUM vs VACUUM FULL"
- 5 steps, eventually correct
- ✗ 3 redundant searches before progress
- ✗ "when to use VACUUM" returned 0 results
- ✗ recovered by guessing doc_id, lucky
→ Phase 2 needs: semantic similarity, reranker

### Q: "How do I configure SSL?"
- 8 steps, hit token limit on fetch
- ✗ libpq-ssl doc is ~30k tokens
→ need chunking, not full-doc fetches

This log is the design pressure for Phase 2. You'll re-read it before writing the retrieval stack.