Beyond Next.js: How to Build a Clean, Production-Grade AI Agent in Python from Scratch

In our previous post, we wrapped up our fourteen-part frontend and full-stack implementation module with a complete Next.js mini-agent project recap, uniting Edge rate limiting, React Server Components, and Supabase persistence into one cohesive workspace. Building inside Next.js and TypeScript is ideal when your agent lives right next to your user interface, but what happens when your agent needs to crunch heavy data tables, run Python-native machine learning models, parse complex PDFs, or operate as an independent backend microservice consumed by multiple apps?

To build a clean, reliable AI agent in Python without heavy framework bloat, you combine Pydantic v2 (BaseModel and Field) for strict tool input validation, httpx with hard timeouts for external I/O, and a bounded while step < max_steps tool-calling loop using the official OpenAI or Anthropic Python SDKs. This architecture mirrors the exact engineering guardrails we built in TypeScript—converting Pydantic schemas directly into LLM tool definitions via .model_json_schema(), catching validation errors gracefully, and recording structured execution traces for every step.

This post launches Section C (Posts 27–36: Backend, Python, Vector Search & DevOps) of our 62-part series. Today we build the standalone Python core of our DevPulse agent so that in Post 28 we can expose it as a high-concurrency FastAPI backend service.

Why Full-Stack JavaScript Developers Still Need Python Agents

If you just spent fourteen posts mastering the Vercel AI SDK, LangChain.js, and Zod in Next.js, you might wonder why you should ever leave TypeScript. In production engineering teams, the choice between a Next.js agent and a Python agent is not ideological—it is driven by workload and ecosystem:

1. Direct Access to the Data & ML Ecosystem: When an agent needs to run numerical regressions, analyze Pandas DataFrames, execute computer vision segmentation, or use local Hugging Face tokenizers and OCR libraries, Python is the native language of those tools.
2. Long-Running Background Workloads: Next.js serverless routes are optimized for fast interactive web requests (typically capped at 10 to 60 seconds). A Python worker running in a Docker container can process a 50-document research queue or run scheduled cron jobs without hitting serverless gateway timeouts.
3. Decoupled Multi-Client Architecture: When your Next.js web app, a Flutter mobile app, and a Telegram or WhatsApp webhook all need to talk to the exact same agent brain, isolating the agent inside a Python backend service keeps your business logic in one place.

Translating Your Mental Model: TypeScript + Zod to Python + Pydantic

The good news is that every architectural rule we learned in Section B translates 1-to-1 into modern Python (3.11+):

• Zod (z.object + .describe()) → Pydantic v2 (BaseModel + Field(description=...)): Just as Zod validates untrusted LLM JSON at runtime in Node.js, Pydantic v2 validates tool arguments in Rust-backed Python and exports standard JSON Schema via .model_json_schema().
• TypeScript ToolResultEnvelope<T> → Python Generic ToolEnvelope Model: Instead of letting a failed Python function raise an unhandled traceback that kills the process, we return a structured { ok: bool, data: Any, agent_instruction: str } envelope.
• fetch() with AbortSignal.timeout(5000) → httpx.Client(timeout=5.0): Python's legacy requests library has no default timeout and blocks forever on hung sockets; modern Python agents use httpx with explicit timeout guards.

Here is how our standalone Python agent orchestrates Pydantic schema validation, LLM reasoning, and tool execution inside a bounded loop:

mermaid flowchart TD Input([User Goal / CLI or API Prompt]) --> Loop{Python Agent Loop current_step <= max_steps 5} Loop -->|Send System Prompt + History + Pydantic JSON Schemas| LLM[Anthropic Claude 3.5 Sonnet / OpenAI GPT-4o] LLM -->|stop_reason == end_turn| Final[Return AgentRunResult final_answer + steps_trace + token_usage] LLM -->|stop_reason == tool_use| Dispatch[Tool Registry Dispatcher] Dispatch --> Validate[Pydantic v2 model_validate Strict Type & Enum Check] Validate -->|ValidationError| ErrEnv[Return Non-Retryable Error Envelope] Validate -->|Valid Args| Exec[Execute Python Tool with httpx 5s Timeout] Exec --> Shaper[DTO Shaper & Text Truncator] Shaper --> Append[Append tool_result Block to Messages] ErrEnv --> Append Append --> Loop Architecture Flow Summary: User Goal → Bounded Python While Loop (max_steps = 5) → LLM with Pydantic JSON Schemas → Pydantic v2 Runtime Validation → httpx Tool Execution & DTO Shaping → Appended Tool Result Observation → Final AgentRunResult.

Step 1: Setting Up a Clean Python Virtual Environment

Never install AI packages globally on your operating system. Create a dedicated project directory (devpulse-python-agent), spin up a virtual environment, and install the official SDKs alongside pydantic, httpx, and python-dotenv:

bash mkdir devpulse-python-agent && cd devpulse-python-agent python3 -m venv .venv source .venv/bin/activate pip install anthropic openai pydantic httpx python-dotenv

Create a .env file in the root of your Python project with your API key:

`ini

.env

ANTHROPIC_API_KEY=sk-ant-your-key-here OPENAI_API_KEY=sk-proj-your-key-here `

Step 2: Defining Pydantic v2 Tool Schemas and Safe Executors (agent/tools.py)

Let us create agent/tools.py. Just like our blueprint.ts and custom-github-tool.ts files in Next.js, this module defines strict input schemas using Pydantic's BaseModel and Field, restricts repositories with Literal allowlists, and wraps live GitHub API lookups inside a 5-second httpx timeout.

`python

agent/tools.py

import json import time from typing import Any, Callable, Dict, Literal, Optional import httpx from pydantic import BaseModel, Field, ValidationError

AllowedRepo = Literal[ 'vercel/next.js', 'vercel/ai', 'langchain-ai/langchainjs', ]

class SearchDocsInput(BaseModel): query: str = Field( min_length=4, max_length=200, description='Concise technical query for searching internal documentation.', ) top_k: int = Field( default=3, ge=1, le=5, description='Number of documentation chunks to retrieve (1 to 5).', )

class InspectGitHubIssueInput(BaseModel): repository: AllowedRepo = Field( description='Allowlisted GitHub repository in owner/name format.' ) issue_number: int = Field( gt=0, le=999999, description='Positive numeric GitHub issue ID (e.g., 402).', )

class DraftTicketInput(BaseModel): title: str = Field(min_length=10, max_length=120) severity: Literal['low', 'medium', 'high', 'critical'] component: Literal['auth', 'api-gateway', 'database', 'frontend-ui'] reproduction_steps: list[str] = Field(min_length=1)

class ToolResultEnvelope(BaseModel): ok: bool tool_name: str retryable: bool duration_ms: int data: Optional[Dict[str, Any]] = None agent_instruction: Optional[str] = None

def execute_search_docs(args: SearchDocsInput) -> Dict[str, Any]: return { 'query_used': args.query, 'top_k': args.top_k, 'matched_chunks': [ { 'doc_path': 'docs/api/rate-limits.md', 'score': 0.93, 'excerpt': ( 'The API gateway enforces a sliding window of 60 requests ' 'per minute per IP. Bursts during token refresh return HTTP 429 ' 'with a Retry-After header.' ), } ], }

def execute_inspect_github_issue(args: InspectGitHubIssueInput) -> Dict[str, Any]: url = f'https://api.github.com/repos/{args.repository}/issues/{args.issue_number}' headers = { 'Accept': 'application/vnd.github+json', 'User-Agent': 'DevPulse-Python-Agent/1.0', }

with httpx.Client(timeout=5.0) as client:
    response = client.get(url, headers=headers)

if response.status_code == 404:
    raise ValueError(
        f'Issue #{args.issue_number} not found in {args.repository}. Do not retry.'
    )

response.raise_for_status()
raw = response.json()
raw_body = raw.get('body') or 'No description provided.'
clean_excerpt = raw_body[:600] + ('...' if len(raw_body) > 600 else '')

labels = [
    lbl.get('name', '') if isinstance(lbl, dict) else str(lbl)
    for lbl in raw.get('labels', [])[:8]
]

return {
    'repository': args.repository,
    'issue_number': raw.get('number'),
    'title': raw.get('title'),
    'state': raw.get('state'),
    'author': (raw.get('user') or {}).get('login', 'unknown'),
    'labels': [l for l in labels if l],
    'comments_count': raw.get('comments', 0),
    'body_excerpt': clean_excerpt,
}

def execute_draft_ticket(args: DraftTicketInput) -> Dict[str, Any]: return { 'draft_id': 'PY-DRAFT-701', 'status': 'awaiting_human_approval', 'ticket_preview': args.model_dump(), }

TOOL_REGISTRY: Dict[str, tuple[type[BaseModel], Callable[[Any], Dict[str, Any]], str]] = { 'search_knowledge_base': ( SearchDocsInput, execute_search_docs, 'Search internal technical documentation for rate limits, auth rules, and architecture specs.', ), 'inspect_live_github_issue': ( InspectGitHubIssueInput, execute_inspect_github_issue, 'Fetch live status, labels, and truncated body for a numeric issue ID in an allowlisted GitHub repo.', ), 'draft_issue_ticket': ( DraftTicketInput, execute_draft_ticket, 'Create a structured bug ticket draft requiring human approval before database insertion.', ), }

def get_anthropic_tool_schemas() -> list[Dict[str, Any]]: schemas = [] for name, (model_cls, _, description) in TOOL_REGISTRY.items(): schemas.append({ 'name': name, 'description': description, 'input_schema': model_cls.model_json_schema(), }) return schemas

def dispatch_agent_tool(tool_name: str, raw_input: Dict[str, Any]) -> str: start_time = time.perf_counter() if tool_name not in TOOL_REGISTRY: envelope = ToolResultEnvelope( ok=False, tool_name=tool_name, retryable=False, duration_ms=0, agent_instruction=f'Unknown tool {tool_name}. Choose a registered tool.', ) return envelope.model_dump_json()

model_cls, executor_fn, _ = TOOL_REGISTRY[tool_name]

try:
    validated_args = model_cls.model_validate(raw_input)
    data = executor_fn(validated_args)
    duration_ms = int((time.perf_counter() - start_time) * 1000)
    envelope = ToolResultEnvelope(
        ok=True,
        tool_name=tool_name,
        retryable=False,
        duration_ms=duration_ms,
        data=data,
    )
    return envelope.model_dump_json()
except ValidationError as val_err:
    duration_ms = int((time.perf_counter() - start_time) * 1000)
    envelope = ToolResultEnvelope(
        ok=False,
        tool_name=tool_name,
        retryable=True,
        duration_ms=duration_ms,
        agent_instruction=f'Pydantic validation failed: {val_err.errors()}. Fix args and retry.',
    )
    return envelope.model_dump_json()
except Exception as exc:
    duration_ms = int((time.perf_counter() - start_time) * 1000)
    envelope = ToolResultEnvelope(
        ok=False,
        tool_name=tool_name,
        retryable=False,
        duration_ms=duration_ms,
        agent_instruction=f'Tool execution error: {str(exc)}',
    )
    return envelope.model_dump_json()

`

Look at how get_anthropic_tool_schemas() uses model_cls.model_json_schema(). In Pydantic v2, you never write raw JSON Schema dictionaries by hand. You write clean Python classes with type hints (Literal, int, str), and Pydantic automatically generates the exact JSON Schema that Claude and OpenAI expect.

Step 3: Writing the Bounded Python Agent Orchestrator (agent/runner.py)

Before reaching for heavy orchestration frameworks like LangGraph or CrewAI, every developer should build a native 60-line agent loop at least once. Writing the while loop yourself removes all the mystery: you see how stop_reason == 'tool_use' triggers your Python function, and how the resulting tool_result block is appended back into the messages list for the next iteration.

Let us create agent/runner.py:

`python

agent/runner.py

import os import time from typing import Any, Dict, List, Optional from anthropic import Anthropic from pydantic import BaseModel from agent.tools import dispatch_agent_tool, get_anthropic_tool_schemas

SYSTEM_PROMPT = ( 'You are DevPulse, a technical triage and documentation AI agent.\n' '1. Always check internal documentation (search_knowledge_base) or live GitHub issues ' '(inspect_live_github_issue) before answering technical questions.\n' '2. Never invent error codes or issue IDs.\n' '3. Only call draft_issue_ticket when asked to log a confirmed bug.' )

class StepTrace(BaseModel): step_index: int thought_summary: str selected_tool: Optional[str] = None tool_input: Optional[Dict[str, Any]] = None tool_output_json: Optional[str] = None duration_ms: int

class AgentRunResult(BaseModel): success: bool final_answer: str steps: List[StepTrace] input_tokens: int output_tokens: int total_duration_ms: int halted_reason: Literal['completed', 'max_steps_reached', 'error']

def run_devpulse_python_agent( user_prompt: str, max_steps: int = 5, model_id: str = 'claude-3-5-sonnet-latest', ) -> AgentRunResult: start_time = time.perf_counter() api_key = os.environ.get('ANTHROPIC_API_KEY') if not api_key: raise RuntimeError('Missing ANTHROPIC_API_KEY in environment.')

client = Anthropic(api_key=api_key)
tools = get_anthropic_tool_schemas()
messages: List[Dict[str, Any]] = [
    {'role': 'user', 'content': user_prompt}
]

steps_trace: List[StepTrace] = []
total_in_tokens = 0
total_out_tokens = 0

for current_step in range(1, max_steps + 1):
    step_start = time.perf_counter()
    response = client.messages.create(
        model=model_id,
        max_tokens=1024,
        temperature=0.1,
        system=SYSTEM_PROMPT,
        tools=tools,
        messages=messages,
    )

    total_in_tokens += response.usage.input_tokens
    total_out_tokens += response.usage.output_tokens

    text_blocks = [
        b.text for b in response.content if getattr(b, 'type', None) == 'text'
    ]
    thought_text = ' '.join(text_blocks).strip() or f'Step {current_step} evaluation'

    if response.stop_reason != 'tool_use':
        steps_trace.append(
            StepTrace(
                step_index=current_step,
                thought_summary=thought_text[:200],
                duration_ms=int((time.perf_counter() - step_start) * 1000),
            )
        )
        return AgentRunResult(
            success=True,
            final_answer=thought_text,
            steps=steps_trace,
            input_tokens=total_in_tokens,
            output_tokens=total_out_tokens,
            total_duration_ms=int((time.perf_counter() - start_time) * 1000),
            halted_reason='completed',
        )

    messages.append({'role': 'assistant', 'content': response.content})
    tool_results_content: List[Dict[str, Any]] = []

    for block in response.content:
        if getattr(block, 'type', None) == 'tool_use':
            tool_name = block.name
            tool_args = block.input if isinstance(block.input, dict) else {}
            tool_id = block.id

            observation_json = dispatch_agent_tool(tool_name, tool_args)

            steps_trace.append(
                StepTrace(
                    step_index=current_step,
                    thought_summary=thought_text[:200],
                    selected_tool=tool_name,
                    tool_input=tool_args,
                    tool_output_json=observation_json,
                    duration_ms=int((time.perf_counter() - step_start) * 1000),
                )
            )

            tool_results_content.append({
                'type': 'tool_result',
                'tool_use_id': tool_id,
                'content': observation_json,
            })

    messages.append({'role': 'user', 'content': tool_results_content})

return AgentRunResult(
    success=False,
    final_answer='Reached maximum reasoning step limit before finalizing answer.',
    steps=steps_trace,
    input_tokens=total_in_tokens,
    output_tokens=total_out_tokens,
    total_duration_ms=int((time.perf_counter() - start_time) * 1000),
    halted_reason='max_steps_reached',
)

`

Look closely at what happens when response.stop_reason == 'tool_use':

1. Appending the Assistant Turn: We append the model's exact response.content (which contains both its intermediate reasoning text and its tool_use block with a unique block.id) to messages.
2. Handling Parallel Tool Calls: Notice the inner for block in response.content loop. If Claude decides to invoke both search_knowledge_base and inspect_live_github_issue in Step 1, our loop executes both tools and collects all tool_result blocks (matched by tool_use_id) into a single user message turn before starting Step 2.
3. Strict Iteration Cap (range(1, max_steps + 1)): Just like maxSteps: 5 in our Next.js agent, the loop cannot run forever.

Step 4: Running and Inspecting Your Python Agent (main.py)

Now create main.py in your project root to load your .env file, run a multi-step triage query, and print the structured Pydantic result:

`python

main.py

from dotenv import load_dotenv from agent.runner import run_devpulse_python_agent

def main() -> None: load_dotenv() query = ( 'Check our internal rate-limit docs for 429 burst rules, ' 'and inspect live GitHub issue #402 in vercel/ai to see its current status.' )

print(f'Running DevPulse Python Agent...\nQuery: {query}\n')
result = run_devpulse_python_agent(user_prompt=query, max_steps=5)

for step in result.steps:
    tool_label = step.selected_tool or 'Final Synthesis'
    print(f'[Step {step.step_index}] {tool_label} ({step.duration_ms}ms)')
    if step.tool_input:
        print(f'  Args: {step.tool_input}')

print('\n--- Final Agent Answer ---')
print(result.final_answer)
print(
    f'\nTelemetry: {result.input_tokens} in / {result.output_tokens} out tokens '
    f'| Total Time: {result.total_duration_ms}ms | Status: {result.halted_reason}'
)

if name == 'main': main() `

Zahid's Engineering Perspective: Why I Stopped Using Untyped Python Dictionaries for Agent Tools

When I was working on Python machine learning pipelines—such as my Gold Price Predictor regression model and deep learning vision experiments—and later started building standalone Python AI agents, I initially fell into the classic Python habit of passing raw dict objects around without runtime schema validation. In my first Python agent script, I wrote a hand-crafted JSON Schema dictionary for the LLM and unpacked the returned tool arguments directly using repo = tool_args['repository'] and limit = int(tool_args['top_k']).

During a multi-step test run, the model hallucinated top_k: 'three' (a string word instead of an integer) and omitted a required filter key. Because there was no Pydantic validation layer sitting between the LLM's raw output and my Python function, int(tool_args['top_k']) threw an unhandled ValueError: invalid literal for int() with base 10: 'three' that crashed the entire Python worker process mid-turn! Even worse, whenever I added a parameter to my Python function, I kept forgetting to update the separate 30-line JSON Schema dictionary.

Moving to Pydantic v2 (BaseModel) solved both headaches immediately. Generating the LLM tool definition directly via model_cls.model_json_schema() guaranteed my prompt schema and Python function signature could never drift apart. And catching ValidationError inside dispatch_agent_tool() to return retryable=True with val_err.errors() allowed the LLM to see its own type mistake and self-correct on the next step without crashing the script.

Common Mistakes When Building AI Agents in Python

As you transition from TypeScript to Python agent development, avoid these 4 critical pitfalls:

1. Using requests.get() Without a Timeout Guard: Python's popular requests library defaults to timeout=None, meaning a single unresponsive external URL will hang your Python worker thread indefinitely. Always use httpx.Client(timeout=5.0) (or httpx.AsyncClient(timeout=5.0) when we move to FastAPI in the next post).
2. Hand-Writing JSON Schemas Instead of Using model_json_schema(): Manually maintaining nested JSON Schema dictionaries in Python is error-prone and leads to silent schema drift. Define a Pydantic BaseModel with Field(description=...) constraints and let Pydantic generate the JSON Schema automatically.
3. Returning Only One tool_result When the LLM Emits Parallel Tool Calls: When Claude or GPT-4o emits two tool_use blocks in a single assistant turn, Anthropic's API strictly requires the very next user message to contain matching tool_result blocks for all emitted tool_use_ids. If you break out of your tool loop after executing only the first tool, the next API call will fail with an HTTP 400 error.
4. Letting Tool Exceptions Crash the for Loop: Just like in Next.js, an external 404 or ValidationError inside a tool should never crash the orchestrator. Always wrap tool execution in a try/except block and return a serialized ToolResultEnvelope(ok=False, agent_instruction=...) string so the model can recover.

Frequently Asked Questions

Why build the Python agent loop with the native SDK + Pydantic first instead of starting with LangChain or CrewAI?

Building the 60-line for current_step in range(1, max_steps + 1) loop with the official SDK and Pydantic v2 teaches you the exact mechanics of stop_reason == 'tool_use', parallel tool_use_id matching, and scratchpad message formatting. Once you understand this core loop, debugging higher-level frameworks like LangGraph, CrewAI, or AutoGen (which we will compare in Section E) becomes effortless—and for many production microservices, this lightweight native loop is all you ever need.

Can I reuse the same Supabase PostgreSQL database from our Next.js project inside this Python agent?

Yes. Because we stored our sessions, messages, and JSONB tool invocations in standard PostgreSQL tables in Supabase back in Post 24, our Python agent can query and write to those exact same tables using the supabase Python package or asyncpg—and in Post 29, we will connect both stacks to Supabase pgvector for semantic search.

How do we turn this standalone Python script into a real web API that our Next.js frontend or mobile app can call?

Right now, run_devpulse_python_agent() runs synchronously in the terminal. In the very next post (Post 28), we will wrap this exact Python agent and its Pydantic schemas inside an asynchronous FastAPI backend server with streaming endpoints, CORS guards, and automatic OpenAPI documentation.