Tools: Building AI Agents with Python: A Practical, Open-Source First Guide

What is an “AI agent” (in practice)? ## Project setup ## Step 1: Define tools (the agent’s capabilities) ## Example tools ## Step 2: Define messages + memory ## Step 3: The model interface (OpenAI-compatible) ## Step 4: Build the agent loop ## Step 5: Try it end-to-end ## Making it more agentic (without making it fragile) ## 1) Add a “planner” step ## 2) Add retrieval (RAG) properly ## 3) Add tool timeouts and cancellation ## 4) Add a strict allowlist and “capabilities” policy ## 5) Add structured tool outputs ## Open-source note: keep the architecture swappable ## Summary AI agents are more than “LLM + prompt.” A useful agent can plan, use tools, remember context, and act safely in the real world (files, APIs, databases). In this post, we’ll build a small but capable agent in Python using an open-source stack. This is aimed at intermediate Python developers who want to understand the moving parts and keep the architecture flexible. A practical agent typically includes: A key design choice: don’t hide the loop. You’ll debug and extend agents more easily when the control flow is visible. Install dependencies: If you’re using an OpenAI-compatible endpoint: Tools are just Python callables plus metadata: We’ll implement a tiny tool framework. We’ll store a basic conversation history plus a “notes” field the agent can update. Many providers (and local gateways) implement an OpenAI-compatible Chat Completions API. We’ll keep this thin so you can swap it out. We’ll ask the model to respond in a structured JSON format: We’ll also add basic guardrails: Create a docs/ folder with a couple .md files (project notes, API docs, etc.). Then run: A typical interaction looks like: Once the basics work, here are practical upgrades. Instead of letting the model decide everything in one shot, add an explicit planning phase: This reduces randomness and improves debuggability. Our search_local_docs is naive substring matching. For real projects, use embeddings: Then create a tool like retrieve_context(query) -> passages. Tools that hit networks should use timeouts: A common mistake is giving agents broad file/network access. Prefer: Returning JSON strings is fine for demos, but you’ll want consistent schemas. Consider: If you later adopt a framework (LangGraph, LlamaIndex, Haystack, Semantic Kernel), you’ll still benefit from understanding: A good rule: frameworks should reduce boilerplate, not hide control flow. You now have a minimal, extensible Python AI agent with: From here, the biggest improvements come from: If you want, I can follow up with a second post that adds: Templates let you quickly answer FAQs or store snippets for re-use. Are you sure you want to hide this comment? It will become hidden in your post, but will still be visible via the comment's permalink. Hide child comments as well For further actions, you may consider blocking this person and/or reporting abuse COMMAND_BLOCK: pip install pydantic httpx Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: pip install pydantic httpx COMMAND_BLOCK: pip install pydantic httpx COMMAND_BLOCK: pip install openai Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: pip install openai COMMAND_BLOCK: pip install openai COMMAND_BLOCK: from __future__ import annotations from dataclasses import dataclass from typing import Any, Callable, Dict, Optional, Type from pydantic import BaseModel, ValidationError @dataclass class Tool: name: str description: str input_model: Type[BaseModel] fn: Callable[..., Any] def run(self, raw_args: Dict[str, Any]) -> Any: args = self.input_model(**raw_args) return self.fn(**args.model_dump()) class ToolRegistry: def __init__(self): self._tools: Dict[str, Tool] = {} def register(self, tool: Tool) -> None: if tool.name in self._tools: raise ValueError(f"Tool already registered: {tool.name}") self._tools[tool.name] = tool def get(self, name: str) -> Tool: return self._tools[name] def list(self) -> Dict[str, Tool]: return dict(self._tools) Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: from __future__ import annotations from dataclasses import dataclass from typing import Any, Callable, Dict, Optional, Type from pydantic import BaseModel, ValidationError @dataclass class Tool: name: str description: str input_model: Type[BaseModel] fn: Callable[..., Any] def run(self, raw_args: Dict[str, Any]) -> Any: args = self.input_model(**raw_args) return self.fn(**args.model_dump()) class ToolRegistry: def __init__(self): self._tools: Dict[str, Tool] = {} def register(self, tool: Tool) -> None: if tool.name in self._tools: raise ValueError(f"Tool already registered: {tool.name}") self._tools[tool.name] = tool def get(self, name: str) -> Tool: return self._tools[name] def list(self) -> Dict[str, Tool]: return dict(self._tools) COMMAND_BLOCK: from __future__ import annotations from dataclasses import dataclass from typing import Any, Callable, Dict, Optional, Type from pydantic import BaseModel, ValidationError @dataclass class Tool: name: str description: str input_model: Type[BaseModel] fn: Callable[..., Any] def run(self, raw_args: Dict[str, Any]) -> Any: args = self.input_model(**raw_args) return self.fn(**args.model_dump()) class ToolRegistry: def __init__(self): self._tools: Dict[str, Tool] = {} def register(self, tool: Tool) -> None: if tool.name in self._tools: raise ValueError(f"Tool already registered: {tool.name}") self._tools[tool.name] = tool def get(self, name: str) -> Tool: return self._tools[name] def list(self) -> Dict[str, Tool]: return dict(self._tools) COMMAND_BLOCK: import os import re from pathlib import Path from typing import List from pydantic import BaseModel, Field class SearchLocalDocsInput(BaseModel): query: str = Field(..., min_length=2) folder: str = Field(..., description="Folder containing .md/.txt files") max_results: int = Field(5, ge=1, le=20) def search_local_docs(query: str, folder: str, max_results: int = 5) -> List[dict]: q = query.lower().strip() folder_path = Path(folder) results = [] for path in folder_path.rglob("*"): if path.suffix.lower() not in {".md", ".txt"}: continue try: text = path.read_text(encoding="utf-8", errors="ignore") except OSError: continue if q in text.lower(): # Grab a small snippet around the first match m = re.search(re.escape(q), text, re.IGNORECASE) start = max(0, m.start() - 120) if m else 0 end = min(len(text), (m.end() + 120) if m else 240) snippet = text[start:end].replace("\n", " ") results.append({"file": str(path), "snippet": snippet}) return results[:max_results] class SummarizeTextInput(BaseModel): text: str = Field(..., min_length=1) max_chars: int = Field(600, ge=100, le=5000) def summarize_text(text: str, max_chars: int = 600) -> str: text = re.sub(r"\s+", " ", text).strip() if len(text) <= max_chars: return text return text[: max_chars - 3] + "..." Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: import os import re from pathlib import Path from typing import List from pydantic import BaseModel, Field class SearchLocalDocsInput(BaseModel): query: str = Field(..., min_length=2) folder: str = Field(..., description="Folder containing .md/.txt files") max_results: int = Field(5, ge=1, le=20) def search_local_docs(query: str, folder: str, max_results: int = 5) -> List[dict]: q = query.lower().strip() folder_path = Path(folder) results = [] for path in folder_path.rglob("*"): if path.suffix.lower() not in {".md", ".txt"}: continue try: text = path.read_text(encoding="utf-8", errors="ignore") except OSError: continue if q in text.lower(): # Grab a small snippet around the first match m = re.search(re.escape(q), text, re.IGNORECASE) start = max(0, m.start() - 120) if m else 0 end = min(len(text), (m.end() + 120) if m else 240) snippet = text[start:end].replace("\n", " ") results.append({"file": str(path), "snippet": snippet}) return results[:max_results] class SummarizeTextInput(BaseModel): text: str = Field(..., min_length=1) max_chars: int = Field(600, ge=100, le=5000) def summarize_text(text: str, max_chars: int = 600) -> str: text = re.sub(r"\s+", " ", text).strip() if len(text) <= max_chars: return text return text[: max_chars - 3] + "..." COMMAND_BLOCK: import os import re from pathlib import Path from typing import List from pydantic import BaseModel, Field class SearchLocalDocsInput(BaseModel): query: str = Field(..., min_length=2) folder: str = Field(..., description="Folder containing .md/.txt files") max_results: int = Field(5, ge=1, le=20) def search_local_docs(query: str, folder: str, max_results: int = 5) -> List[dict]: q = query.lower().strip() folder_path = Path(folder) results = [] for path in folder_path.rglob("*"): if path.suffix.lower() not in {".md", ".txt"}: continue try: text = path.read_text(encoding="utf-8", errors="ignore") except OSError: continue if q in text.lower(): # Grab a small snippet around the first match m = re.search(re.escape(q), text, re.IGNORECASE) start = max(0, m.start() - 120) if m else 0 end = min(len(text), (m.end() + 120) if m else 240) snippet = text[start:end].replace("\n", " ") results.append({"file": str(path), "snippet": snippet}) return results[:max_results] class SummarizeTextInput(BaseModel): text: str = Field(..., min_length=1) max_chars: int = Field(600, ge=100, le=5000) def summarize_text(text: str, max_chars: int = 600) -> str: text = re.sub(r"\s+", " ", text).strip() if len(text) <= max_chars: return text return text[: max_chars - 3] + "..." CODE_BLOCK: registry = ToolRegistry() registry.register( Tool( name="search_local_docs", description="Search local markdown/text files for a query and return file snippets.", input_model=SearchLocalDocsInput, fn=search_local_docs, ) ) registry.register( Tool( name="summarize_text", description="Summarize text by truncating to a max character length.", input_model=SummarizeTextInput, fn=summarize_text, ) ) Enter fullscreen mode Exit fullscreen mode CODE_BLOCK: registry = ToolRegistry() registry.register( Tool( name="search_local_docs", description="Search local markdown/text files for a query and return file snippets.", input_model=SearchLocalDocsInput, fn=search_local_docs, ) ) registry.register( Tool( name="summarize_text", description="Summarize text by truncating to a max character length.", input_model=SummarizeTextInput, fn=summarize_text, ) ) CODE_BLOCK: registry = ToolRegistry() registry.register( Tool( name="search_local_docs", description="Search local markdown/text files for a query and return file snippets.", input_model=SearchLocalDocsInput, fn=search_local_docs, ) ) registry.register( Tool( name="summarize_text", description="Summarize text by truncating to a max character length.", input_model=SummarizeTextInput, fn=summarize_text, ) ) COMMAND_BLOCK: from dataclasses import dataclass, field from typing import Literal, List Role = Literal["system", "user", "assistant", "tool"] @dataclass class Message: role: Role content: str name: str | None = None # used for tool name @dataclass class AgentState: messages: List[Message] = field(default_factory=list) notes: str = "" def add(self, role: Role, content: str, name: str | None = None) -> None: self.messages.append(Message(role=role, content=content, name=name)) Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: from dataclasses import dataclass, field from typing import Literal, List Role = Literal["system", "user", "assistant", "tool"] @dataclass class Message: role: Role content: str name: str | None = None # used for tool name @dataclass class AgentState: messages: List[Message] = field(default_factory=list) notes: str = "" def add(self, role: Role, content: str, name: str | None = None) -> None: self.messages.append(Message(role=role, content=content, name=name)) COMMAND_BLOCK: from dataclasses import dataclass, field from typing import Literal, List Role = Literal["system", "user", "assistant", "tool"] @dataclass class Message: role: Role content: str name: str | None = None # used for tool name @dataclass class AgentState: messages: List[Message] = field(default_factory=list) notes: str = "" def add(self, role: Role, content: str, name: str | None = None) -> None: self.messages.append(Message(role=role, content=content, name=name)) COMMAND_BLOCK: import json from typing import Any, Dict class LLMClient: def __init__(self, model: str = "gpt-4o-mini"): from openai import OpenAI self.client = OpenAI() self.model = model def chat(self, messages: list[dict]) -> str: resp = self.client.chat.completions.create( model=self.model, messages=messages, temperature=0.2, ) return resp.choices[0].message.content def to_openai_messages(state: AgentState) -> list[dict]: msgs = [] for m in state.messages: d = {"role": m.role, "content": m.content} if m.name: d["name"] = m.name msgs.append(d) return msgs Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: import json from typing import Any, Dict class LLMClient: def __init__(self, model: str = "gpt-4o-mini"): from openai import OpenAI self.client = OpenAI() self.model = model def chat(self, messages: list[dict]) -> str: resp = self.client.chat.completions.create( model=self.model, messages=messages, temperature=0.2, ) return resp.choices[0].message.content def to_openai_messages(state: AgentState) -> list[dict]: msgs = [] for m in state.messages: d = {"role": m.role, "content": m.content} if m.name: d["name"] = m.name msgs.append(d) return msgs COMMAND_BLOCK: import json from typing import Any, Dict class LLMClient: def __init__(self, model: str = "gpt-4o-mini"): from openai import OpenAI self.client = OpenAI() self.model = model def chat(self, messages: list[dict]) -> str: resp = self.client.chat.completions.create( model=self.model, messages=messages, temperature=0.2, ) return resp.choices[0].message.content def to_openai_messages(state: AgentState) -> list[dict]: msgs = [] for m in state.messages: d = {"role": m.role, "content": m.content} if m.name: d["name"] = m.name msgs.append(d) return msgs COMMAND_BLOCK: SYSTEM_PROMPT = """ You are a helpful AI agent. You can either: 1) Call a tool, by responding with strict JSON: {"type":"tool","name":"...","args":{...}} 2) Or answer the user, by responding with strict JSON: {"type":"final","answer":"..."} Rules: - Only call tools that are available. - If you call a tool, keep args minimal and valid. - Use the agent notes when helpful. - Output MUST be valid JSON and nothing else. """.strip() class Agent: def __init__(self, llm: LLMClient, tools: ToolRegistry): self.llm = llm self.tools = tools def run(self, user_input: str, state: Optional[AgentState] = None, max_steps: int = 8) -> str: state = state or AgentState() # Add system prompt once at the start if not state.messages or state.messages[0].role != "system": state.messages.insert(0, Message("system", SYSTEM_PROMPT)) state.add("user", user_input) for step in range(max_steps): # Provide notes as context (simple approach) if state.notes: state.add("system", f"Agent notes: {state.notes}") raw = self.llm.chat(to_openai_messages(state)) try: payload = json.loads(raw) except json.JSONDecodeError: # If the model misbehaves, force a final response return "Model returned non-JSON output. Try again with a stricter prompt." if payload.get("type") == "final": answer = payload.get("answer", "") state.add("assistant", answer) return answer if payload.get("type") == "tool": name = payload.get("name") args = payload.get("args") or {} if name not in self.tools.list(): state.add("tool", f"ERROR: tool not allowed: {name}", name=name) continue tool = self.tools.get(name) try: result = tool.run(args) state.add("tool", json.dumps(result, ensure_ascii=False), name=name) except ValidationError as ve: state.add("tool", f"VALIDATION_ERROR: {ve}", name=name) except Exception as e: state.add("tool", f"TOOL_ERROR: {e}", name=name) continue # Unknown response type state.add("assistant", "I couldn't determine the next action.") return "I couldn't determine the next action." return "Max steps reached without a final answer." Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: SYSTEM_PROMPT = """ You are a helpful AI agent. You can either: 1) Call a tool, by responding with strict JSON: {"type":"tool","name":"...","args":{...}} 2) Or answer the user, by responding with strict JSON: {"type":"final","answer":"..."} Rules: - Only call tools that are available. - If you call a tool, keep args minimal and valid. - Use the agent notes when helpful. - Output MUST be valid JSON and nothing else. """.strip() class Agent: def __init__(self, llm: LLMClient, tools: ToolRegistry): self.llm = llm self.tools = tools def run(self, user_input: str, state: Optional[AgentState] = None, max_steps: int = 8) -> str: state = state or AgentState() # Add system prompt once at the start if not state.messages or state.messages[0].role != "system": state.messages.insert(0, Message("system", SYSTEM_PROMPT)) state.add("user", user_input) for step in range(max_steps): # Provide notes as context (simple approach) if state.notes: state.add("system", f"Agent notes: {state.notes}") raw = self.llm.chat(to_openai_messages(state)) try: payload = json.loads(raw) except json.JSONDecodeError: # If the model misbehaves, force a final response return "Model returned non-JSON output. Try again with a stricter prompt." if payload.get("type") == "final": answer = payload.get("answer", "") state.add("assistant", answer) return answer if payload.get("type") == "tool": name = payload.get("name") args = payload.get("args") or {} if name not in self.tools.list(): state.add("tool", f"ERROR: tool not allowed: {name}", name=name) continue tool = self.tools.get(name) try: result = tool.run(args) state.add("tool", json.dumps(result, ensure_ascii=False), name=name) except ValidationError as ve: state.add("tool", f"VALIDATION_ERROR: {ve}", name=name) except Exception as e: state.add("tool", f"TOOL_ERROR: {e}", name=name) continue # Unknown response type state.add("assistant", "I couldn't determine the next action.") return "I couldn't determine the next action." return "Max steps reached without a final answer." COMMAND_BLOCK: SYSTEM_PROMPT = """ You are a helpful AI agent. You can either: 1) Call a tool, by responding with strict JSON: {"type":"tool","name":"...","args":{...}} 2) Or answer the user, by responding with strict JSON: {"type":"final","answer":"..."} Rules: - Only call tools that are available. - If you call a tool, keep args minimal and valid. - Use the agent notes when helpful. - Output MUST be valid JSON and nothing else. """.strip() class Agent: def __init__(self, llm: LLMClient, tools: ToolRegistry): self.llm = llm self.tools = tools def run(self, user_input: str, state: Optional[AgentState] = None, max_steps: int = 8) -> str: state = state or AgentState() # Add system prompt once at the start if not state.messages or state.messages[0].role != "system": state.messages.insert(0, Message("system", SYSTEM_PROMPT)) state.add("user", user_input) for step in range(max_steps): # Provide notes as context (simple approach) if state.notes: state.add("system", f"Agent notes: {state.notes}") raw = self.llm.chat(to_openai_messages(state)) try: payload = json.loads(raw) except json.JSONDecodeError: # If the model misbehaves, force a final response return "Model returned non-JSON output. Try again with a stricter prompt." if payload.get("type") == "final": answer = payload.get("answer", "") state.add("assistant", answer) return answer if payload.get("type") == "tool": name = payload.get("name") args = payload.get("args") or {} if name not in self.tools.list(): state.add("tool", f"ERROR: tool not allowed: {name}", name=name) continue tool = self.tools.get(name) try: result = tool.run(args) state.add("tool", json.dumps(result, ensure_ascii=False), name=name) except ValidationError as ve: state.add("tool", f"VALIDATION_ERROR: {ve}", name=name) except Exception as e: state.add("tool", f"TOOL_ERROR: {e}", name=name) continue # Unknown response type state.add("assistant", "I couldn't determine the next action.") return "I couldn't determine the next action." return "Max steps reached without a final answer." CODE_BLOCK: if __name__ == "__main__": llm = LLMClient(model="gpt-4o-mini") agent = Agent(llm=llm, tools=registry) question = "Search my docs for 'rate limit' and explain what it says in 3 bullet points. Folder is docs." print(agent.run(question)) Enter fullscreen mode Exit fullscreen mode CODE_BLOCK: if __name__ == "__main__": llm = LLMClient(model="gpt-4o-mini") agent = Agent(llm=llm, tools=registry) question = "Search my docs for 'rate limit' and explain what it says in 3 bullet points. Folder is docs." print(agent.run(question)) CODE_BLOCK: if __name__ == "__main__": llm = LLMClient(model="gpt-4o-mini") agent = Agent(llm=llm, tools=registry) question = "Search my docs for 'rate limit' and explain what it says in 3 bullet points. Folder is docs." print(agent.run(question)) COMMAND_BLOCK: import httpx def fetch_url(url: str) -> str: with httpx.Client(timeout=10.0, follow_redirects=True) as client: return client.get(url).text Enter fullscreen mode Exit fullscreen mode COMMAND_BLOCK: import httpx def fetch_url(url: str) -> str: with httpx.Client(timeout=10.0, follow_redirects=True) as client: return client.get(url).text COMMAND_BLOCK: import httpx def fetch_url(url: str) -> str: with httpx.Client(timeout=10.0, follow_redirects=True) as client: return client.get(url).text - A minimal agent loop (think/plan → tool call → observe → repeat) - A tool registry with typed inputs - Lightweight memory (conversation + notes) - Basic guardrails (tool allowlist + timeouts + validation) - A working example: an agent that can search docs (locally), summarize, and draft a response - Model: an LLM that can reason over text and choose actions. - Tools: functions the model can call (HTTP requests, DB queries, file I/O). - Memory: state across turns (chat history, scratchpad, retrieved notes). - Policy/Loop: logic that decides when to call tools and when to stop. - Safety: constraints to avoid dangerous actions. - Python 3.11+ - pydantic for tool input validation - httpx (optional) for web calls - An LLM client (examples include OpenAI-compatible APIs or local models). I’ll show an OpenAI-compatible interface, but the agent architecture is model-agnostic. - Description (for the model) - Input schema - Function to execute - search_local_docs: search a local folder of markdown/text files - summarize_text: a non-LLM “tool” (simple chunking + truncation) to show that tools can be deterministic - Either: { "type": "final", "answer": "..." } - Or: { "type": "tool", "name": "...", "args": { ... } } - Send system prompt + history + notes - Parse model output - If tool call: validate args, run tool, append tool result - If final: return answer - Stop after N steps - Tool allowlist: only registered tools can run - Validation: Pydantic schemas - Step limit: prevents infinite loops - Model calls search_local_docs with {query: "rate limit", folder: "docs"} - Tool returns snippets - Model calls summarize_text (optional) - Model returns a final bullet list - Step A: produce a plan (no tools) - Step B: execute the next tool call - sentence-transformers for local embeddings - A vector store like FAISS, Chroma, or SQLite-based solutions - A small set of tools - Explicit path sandboxing (only within a workspace directory) - Read-only tools by default - Tool output models (Pydantic) - A standardized envelope: { ok: bool, data: ..., error: ... } - How tools are validated - Where memory lives - How the loop terminates - How errors are handled - A clear agent loop - Typed tools with validation - Basic memory - Guardrails (allowlist, step limit) - Better retrieval (embeddings) - Better planning (explicit plan/execute) - Better safety (sandboxing + permissions) - Embeddings + FAISS for retrieval - A planner/executor split - Streaming outputs and better tracing/logging