Building Your First Agent — AI Tutorials

The Project: Research Assistant Agent

We'll build a Research Assistant Agent that:

Accepts a research topic from the user
Searches the web for recent, relevant information
Extracts and synthesizes key insights
Maintains a persistent memory of past research sessions
Produces a structured Markdown report
Saves the report to disk

This is a practical, complete system — not a toy example.

Step 1: Define the Interface

Before writing any agent code, define the interface:

# What does the agent accept?
def research(
    topic: str,
    depth: Literal["quick", "deep"] = "quick",  # 3 or 10 searches
    output_path: str | None = None,              # save report here
) -> ResearchReport:
    ...

# What does it return?
@dataclass
class ResearchReport:
    topic: str
    summary: str           # 2-3 sentence TL;DR
    key_findings: list[str]  # bullet points
    sections: dict[str, str] # heading -> content
    sources: list[Source]    # cited URLs
    created_at: datetime

This interface-first approach forces you to think about what success looks like before you write a single LLM call.

Step 2: Design the Tools

Tool Definitions with Full Schemaspython

from typing import Literal
from dataclasses import dataclass, field
from datetime import datetime
import json, os, re, hashlib

# Tool schemas — the LLM sees these descriptions and parameters
TOOLS = [
    {
        "name": "web_search",
        "description": (
            "Search the internet for current information on a topic. "
            "Returns titles, URLs, and snippets. "
            "Prefer specific queries over vague ones for better results."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "Specific search query (max 100 chars)"
                },
                "num_results": {
                    "type": "integer",
                    "description": "Number of results to return (1-10)",
                    "default": 5
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "fetch_webpage",
        "description": (
            "Fetch and extract the text content of a webpage URL. "
            "Use this to read full articles after finding promising URLs via search."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "url": {"type": "string", "description": "Full URL to fetch"},
                "max_chars": {
                    "type": "integer",
                    "description": "Maximum characters to return (default 5000)",
                    "default": 5000
                }
            },
            "required": ["url"]
        }
    },
    {
        "name": "save_note",
        "description": (
            "Save a note or finding to the research memory. "
            "Use this to preserve important facts before they scroll out of context."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "key": {"type": "string", "description": "Short label for this note"},
                "content": {"type": "string", "description": "The note content"},
                "source_url": {
                    "type": "string",
                    "description": "URL where this fact was found (if applicable)"
                }
            },
            "required": ["key", "content"]
        }
    },
    {
        "name": "recall_notes",
        "description": "Retrieve all saved research notes from this session.",
        "input_schema": {
            "type": "object",
            "properties": {}
        }
    },
    {
        "name": "write_report",
        "description": (
            "Write the final research report. Call this when you have gathered "
            "sufficient information and are ready to synthesize findings."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "summary": {"type": "string", "description": "2-3 sentence executive summary"},
                "key_findings": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "5-8 bullet points of key findings"
                },
                "sections": {
                    "type": "object",
                    "description": "Section headings mapped to content (3-5 sections)"
                },
                "sources": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "title": {"type": "string"},
                            "url": {"type": "string"}
                        }
                    },
                    "description": "List of sources cited in the report"
                }
            },
            "required": ["summary", "key_findings", "sections", "sources"]
        }
    }
]

Tool Implementationspython

import requests
from bs4 import BeautifulSoup

class ToolExecutor:
    def __init__(self):
        self.notes: dict[str, dict] = {}
        self.final_report: dict | None = None

    def execute(self, name: str, args: dict) -> str:
        handler = getattr(self, f"_tool_{name}", None)
        if not handler:
            return f"Error: unknown tool '{name}'"
        try:
            return handler(**args)
        except Exception as e:
            return f"Tool error in {name}: {type(e).__name__}: {e}"

    def _tool_web_search(self, query: str, num_results: int = 5) -> str:
        """Use a real search API in production (Tavily, SerpAPI, Brave)."""
        # Mock implementation — replace with:
        # from tavily import TavilyClient
        # client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
        # results = client.search(query, max_results=num_results)
        return json.dumps([
            {
                "title": f"Result {i}: {query}",
                "url": f"https://example.com/result-{i}",
                "snippet": f"This article covers {query} in depth..."
            }
            for i in range(1, num_results + 1)
        ], indent=2)

    def _tool_fetch_webpage(self, url: str, max_chars: int = 5000) -> str:
        try:
            resp = requests.get(url, timeout=10, headers={"User-Agent": "ResearchBot/1.0"})
            resp.raise_for_status()
            soup = BeautifulSoup(resp.text, "html.parser")
            # Remove scripts and styles
            for tag in soup(["script", "style", "nav", "footer"]):
                tag.decompose()
            text = soup.get_text(separator="\n", strip=True)
            return text[:max_chars]
        except Exception as e:
            return f"Failed to fetch {url}: {e}"

    def _tool_save_note(self, key: str, content: str, source_url: str = "") -> str:
        self.notes[key] = {
            "content": content,
            "source_url": source_url,
            "saved_at": datetime.now().isoformat()
        }
        return f"Note saved: '{key}' ({len(content)} chars)"

    def _tool_recall_notes(self) -> str:
        if not self.notes:
            return "No notes saved yet."
        return json.dumps(self.notes, indent=2)

    def _tool_write_report(self, summary: str, key_findings: list,
                            sections: dict, sources: list) -> str:
        self.final_report = {
            "summary": summary,
            "key_findings": key_findings,
            "sections": sections,
            "sources": sources,
        }
        return "REPORT_COMPLETE"  # signals the agent to stop

The Complete Agent Looppython

from anthropic import Anthropic

client = Anthropic()

SYSTEM_PROMPT = """You are a meticulous research assistant. When given a research topic:

1. Search for 3-5 different angles or subtopics
2. Fetch and read at least 2-3 full articles
3. Save important notes as you go (don't rely solely on context)
4. When you have sufficient information, write a structured report

Research depth guide:
- "quick": 3 searches, 2 articles, 5-6 findings
- "deep": 8-10 searches, 5-6 articles, 8-10 findings

Always cite sources. Prefer recent information (last 12 months when possible).
When you call write_report, use REPORT_COMPLETE in the response to signal completion."""

def run_research_agent(
    topic: str,
    depth: str = "quick",
    output_path: str | None = None,
    max_steps: int = 20,
) -> dict:
    executor = ToolExecutor()
    messages = [{
        "role": "user",
        "content": f"Research this topic in {depth} depth: {topic}"
    }]

    step = 0
    while step < max_steps:
        step += 1
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            system=SYSTEM_PROMPT,
            tools=TOOLS,
            messages=messages,
        )

        # Check if agent is done (either no more tools, or wrote the report)
        messages.append({"role": "assistant", "content": response.content})

        tool_results = []
        report_complete = False

        for block in response.content:
            if block.type == "tool_use":
                result = executor.execute(block.name, block.input)
                print(f"  [{step}] {block.name}({list(block.input.keys())}) -> {result[:80]}...")
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                })
                if result == "REPORT_COMPLETE":
                    report_complete = True

        if response.stop_reason == "end_turn" and not tool_results:
            break

        if tool_results:
            messages.append({"role": "user", "content": tool_results})

        if report_complete:
            break

    report = executor.final_report
    if report and output_path:
        _save_report_markdown(topic, report, output_path)

    return report or {"error": "Agent did not produce a report"}

def _save_report_markdown(topic: str, report: dict, path: str) -> None:
    lines = [f"# Research Report: {topic}\n"]
    lines.append(f"## Summary\n{report['summary']}\n")
    lines.append("## Key Findings\n")
    for finding in report["key_findings"]:
        lines.append(f"- {finding}")
    lines.append("")
    for heading, content in report.get("sections", {}).items():
        lines.append(f"\n## {heading}\n{content}")
    lines.append("\n## Sources\n")
    for src in report.get("sources", []):
        lines.append(f"- [{src['title']}]({src['url']})")
    with open(path, "w") as f:
        f.write("\n".join(lines))
    print(f"Report saved to {path}")

# Run it
if __name__ == "__main__":
    report = run_research_agent(
        topic="Model Context Protocol (MCP) by Anthropic",
        depth="quick",
        output_path="report.md",
    )
    print("\n=== REPORT SUMMARY ===")
    print(report.get("summary", "No summary"))

Step 3: Evaluating Your Agent

The hardest part of building agents isn't the code — it's knowing when they work reliably enough to ship. Build an evaluation harness early.

Agent Evaluation Harnesspython

import pytest
from unittest.mock import patch

# Test cases: input + expected behavior
TEST_CASES = [
    {
        "topic": "Python asyncio event loop",
        "depth": "quick",
        "checks": [
            lambda r: len(r.get("key_findings", [])) >= 4,
            lambda r: len(r.get("sections", {})) >= 2,
            lambda r: len(r.get("sources", [])) >= 1,
            lambda r: "asyncio" in r.get("summary", "").lower(),
        ],
        "labels": [
            "has >= 4 findings",
            "has >= 2 sections",
            "has at least 1 source",
            "summary mentions asyncio",
        ]
    },
    {
        "topic": "Quantum computing 2024",
        "depth": "quick",
        "checks": [
            lambda r: r.get("summary"),
            lambda r: "error" not in r,
        ],
        "labels": ["produces summary", "no error"]
    }
]

def evaluate_agent(n_runs: int = 3) -> dict:
    """Run evaluation across test cases and return pass rates."""
    results = []

    for test in TEST_CASES:
        passes = []
        for run in range(n_runs):
            report = run_research_agent(test["topic"], test["depth"])
            run_passes = []
            for check, label in zip(test["checks"], test["labels"]):
                passed = check(report)
                run_passes.append({"check": label, "passed": passed})
                if not passed:
                    print(f"FAIL [{test['topic']}] {label}")
            passes.append(run_passes)

        # Compute pass rate per check
        for i, label in enumerate(test["labels"]):
            pass_rate = sum(1 for run in passes if run[i]["passed"]) / n_runs
            results.append({
                "topic": test["topic"],
                "check": label,
                "pass_rate": pass_rate,
            })

    return results

# Run eval
if __name__ == "__main__":
    results = evaluate_agent(n_runs=3)
    for r in results:
        icon = "✓" if r["pass_rate"] >= 0.8 else "✗"
        print(f"{icon} {r['topic'][:30]} | {r['check']}: {r['pass_rate']:.0%}")

Step 4: Production Best Practices

Structured Logging

Every tool call, LLM call, and decision should be logged:

import structlog

log = structlog.get_logger()

# In your agent loop
log.info("tool_call", tool=block.name, args=block.input, step=step)
log.info("tool_result", tool=block.name, result_len=len(result), step=step)
log.info("agent_complete", topic=topic, steps=step, report_sections=len(report.get("sections", {})))

Retry Logic for Tool Failures

import tenacity

@tenacity.retry(
    stop=tenacity.stop_after_attempt(3),
    wait=tenacity.wait_exponential(multiplier=1, min=2, max=10),
    retry=tenacity.retry_if_exception_type(requests.RequestException),
)
def fetch_with_retry(url: str) -> str:
    return _tool_fetch_webpage(url)

Graceful Degradation

def _tool_web_search(self, query: str, num_results: int = 5) -> str:
    for api in [self._search_tavily, self._search_brave, self._search_serpapi]:
        try:
            return api(query, num_results)
        except Exception as e:
            log.warning("search_api_failed", api=api.__name__, error=str(e))
    return json.dumps([])  # return empty results, don't crash the agent

Rate Limiting

from ratelimit import limits, sleep_and_retry

@sleep_and_retry
@limits(calls=5, period=1)  # max 5 calls per second
def rate_limited_search(query: str) -> str:
    return _tool_web_search(query)

Deployment Checklist

Before shipping your agent to production:

Category	Check
Safety	Tool actions are reversible or require human approval
Reliability	Retry logic for all external API calls
Limits	Max iterations, timeouts on tool calls
Observability	Structured logs with trace IDs
Evaluation	>80% pass rate on test suite
Secrets	No hardcoded API keys (use env vars)
Cost	Token budget set per run
Errors	Graceful failure messages, no crashes

Knowledge check

Why is it important to define the agent's output interface BEFORE writing the agent loop?

Congratulations — You've Completed the Agentic AI Module!

You've covered the full spectrum of agentic AI:

Introduction — What agents are and why they matter
Architectures — ReAct, tool calling, memory patterns
CrewAI — Role-based multi-agent collaboration
LangGraph — Stateful, graph-based workflows
AutoGen — Conversational multi-agent systems
MCP — The open standard for tool integration
Use Cases — Real production deployments
Building Your First Agent — End-to-end implementation

What's Next?

Explore the KG + Agents intersection: combine knowledge graphs with agents for structured reasoning over connected data
Try the frameworks: start with a CrewAI tutorial or the LangGraph quickstart
Join the community: r/MachineLearning, Hugging Face Discord, LangChain Discord
Read the papers: "ReAct: Synergizing Reasoning and Acting in Language Models" (Yao et al., 2022)

The agentic AI space is moving fast. The best way to stay current is to build things.

← Agentic AI