AI Agent Development and Interview Guide: 01-Python Async Backend and Engineering Fundamentals

1. Synchronous, Concurrency, and Parallelism

First distinguish three concepts:

• Synchronous execution: Next step executes only after current call finishes. This is a critical execution standard and core baseline in production engineering architecture and system design.
• Concurrency: Multiple tasks progress alternately within the same time slice, ideal for extensive I/O waiting. This is a critical execution standard and core baseline in production engineering architecture and system design.
• Parallelism: Multiple CPU cores compute simultaneously, ideal for CPU-intensive tasks. This is a critical execution standard and core baseline in production engineering architecture and system design.

Model API requests are generally I/O-bound. Embedding batching is also I/O when calling remote services; if running models locally, it may be CPU/GPU-bound. asyncio can reduce waiting waste, but will not automatically accelerate local matrix computations.

FastAPI official recommendation: Use async def and await when third-party libraries provide awaitable interfaces; blocking libraries should be placed in normal def routes or thread pools to avoid blocking the event loop. Refer to FastAPI async documentation.

🔥 P0 High-Frequency Essential: Is async always faster? No. Asynchrony improves concurrency utilization during I/O wait times. CPU-intensive tasks still require multiprocessing, native parallel libraries, GPUs, or task queues. Placing blocking functions directly into async def will instead freeze the entire event loop.

2. A Concurrency-Controlled Model Invoker

The example below demonstrates four production-critical points: concurrency limits, timeouts, retries, and structured errors.

import asyncio
import random
from dataclasses import dataclass
from typing import Protocol


class ModelClient(Protocol):
    async def generate(self, prompt: str) -> str: ...


@dataclass
class ModelCallError(Exception):
    code: str
    message: str
    retryable: bool


class ReliableModelGateway:
    def __init__(
        self,
        client: ModelClient,
        max_concurrency: int = 8,
        timeout_seconds: float = 20.0,
        max_attempts: int = 3,
    ) -> None:
        self.client = client
        self.semaphore = asyncio.Semaphore(max_concurrency)
        self.timeout_seconds = timeout_seconds
        self.max_attempts = max_attempts

    async def generate(self, prompt: str) -> str:
        async with self.semaphore:
            for attempt in range(1, self.max_attempts + 1):
                try:
                    async with asyncio.timeout(self.timeout_seconds):
                        return await self.client.generate(prompt)
                except TimeoutError:
                    error = ModelCallError(
                        code="MODEL_TIMEOUT",
                        message=f"model timed out after {self.timeout_seconds}s",
                        retryable=True,
                    )
                except ConnectionError as exc:
                    error = ModelCallError(
                        code="MODEL_NETWORK_ERROR",
                        message=str(exc),
                        retryable=True,
                    )
                except ValueError as exc:
                    raise ModelCallError(
                        code="MODEL_BAD_REQUEST",
                        message=str(exc),
                        retryable=False,
                    ) from exc

                if attempt == self.max_attempts or not error.retryable:
                    raise error

                base = 0.25 * (2 ** (attempt - 1))
                await asyncio.sleep(base + random.uniform(0, 0.1))

        raise RuntimeError("unreachable")

Several points often overlooked here:

• Semaphore limits the number of requests entering external model calls simultaneously, preventing triggering rate limits or exhausting connection pools.
• Timeouts must be configured on the caller side; never assume providers will always return in a timely manner.
• Only retry transient errors. Invalid parameters, insufficient permissions, and content policy rejections should generally not be blindly retried.
• Retries amplify downstream pressure, and thus require maximum attempt caps, backoffs, and end-to-end deadline budgets.

🔥 P0 High-Frequency Essential: Why combine exponential backoff with random jitter? Fixed intervals cause concurrently failing requests to hammer downstream services again at the exact same moment. Exponential backoff decreases retry frequency, while random jitter disperses retry timing. Answers should also mention maximum retry counts, retrying only idempotent operations, and global timeout budgets.

3. FastAPI Service Boundaries

The API layer should not be directly stuffed with Agent logic. A 4-layer clean architecture is recommended:

HTTP / API Layer -> Application Orchestration Layer -> Domain / Tool Layer -> Infrastructure Adapter Layer

• API layer: Authentication, parameter validation, request IDs, HTTP status codes.
• Application layer: Starting Agent Runs, reading state, canceling tasks.
• Domain layer: Business rules, tool permissions, approval conditions.
• Infrastructure layer: Concrete clients for models, databases, vector stores, queues.

A minimal interface:

from typing import Literal
from uuid import UUID, uuid4

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, Field

app = FastAPI()


class RunRequest(BaseModel):
    query: str = Field(min_length=1, max_length=4000)
    mode: Literal["read_only", "allow_write"] = "read_only"


class RunAccepted(BaseModel):
    run_id: UUID
    status: Literal["accepted"] = "accepted"


@app.post("/v1/runs", response_model=RunAccepted, status_code=202)
async def create_run(
    body: RunRequest,
    idempotency_key: str = Header(min_length=8),
) -> RunAccepted:
    if body.mode == "allow_write" and not idempotency_key:
        raise HTTPException(status_code=400, detail="missing idempotency key")

    run_id = uuid4()
#    # Push long-running tasks to a queue; do not keep HTTP requests waiting indefinitely.
    return RunAccepted(run_id=run_id)

Why Return 202

Long-running Agent Runs may take tens of seconds or even minutes. Synchronous waiting occupies connections and makes retry semantics complex. Common practice is:

• POST /runs returns 202 Accepted and run_id;
• Background Worker executes;
• Client obtains progress via polling, SSE, or WebSocket;
• Supports cancellation, timeouts, and human approval.

Short, stable, read-only requests can return synchronously; complex tasks are better suited for asynchronous task models.

4. Idempotency: One of the Most Critical Backend Concepts for Agent Tools

If network timeouts occur, the caller does not know whether "creating a ticket" actually succeeded. Directly retrying may create duplicate tickets. The solution is ensuring that executing identical business requests multiple times produces the effect only once.

from dataclasses import dataclass


@dataclass
class CreateTicketCommand:
    idempotency_key: str
    user_id: str
    title: str
    description: str


async def create_ticket(command: CreateTicketCommand, repository) -> str:
    existing = await repository.find_by_idempotency_key(command.idempotency_key)
    if existing:
        return existing.ticket_id

#    # Add a unique index on idempotency_key in the database to prevent concurrent duplicate writes.
    ticket = await repository.insert(command)
    return ticket.ticket_id

Truly reliable implementations also require database unique constraints or transactions; relying solely on "check-then-write" creates concurrency race conditions.

🔥 P0 High-Frequency Essential: What is the relationship between retries and idempotency? Retries recover transient failures, while idempotency ensures retries do not repeatedly produce side effects. Read-only operations are naturally closer to idempotent; write operations require idempotency keys, unique constraints, state machines, or compensating transactions.

5. Do Not Block the Event Loop

Anti-pattern example:

import time

async def bad_handler():
    time.sleep(5)  # Blocks the entire event loop

If calling synchronous blocking libraries is unavoidable, temporarily offload them to threads:

import asyncio

async def parse_pdf(path: str) -> str:
    return await asyncio.to_thread(blocking_pdf_parser, path)

However, thread pools are not infinite resources. Heavy document parsing, OCR, or local inference should be placed in standalone Workers/queues configured with concurrency caps.

6. Error Classification and Recovery Strategies

Do not convert all exceptions into raw strings and feed them blindly back to the model. Classify first, then decide between system retry, model correction, user intervention, or direct failure.

7. Testing an Agent Tool

Never invoke real external systems during automated testing. Use dependency injection to pass Fakes:

import pytest


class FakeRepository:
    def __init__(self):
        self.items = {}

    async def find_by_idempotency_key(self, key):
        return self.items.get(key)

    async def insert(self, command):
        ticket = type("Ticket", (), {"ticket_id": "T-001"})()
        self.items[command.idempotency_key] = ticket
        return ticket


@pytest.mark.asyncio
async def test_create_ticket_is_idempotent():
    repo = FakeRepository()
    command = CreateTicketCommand("req-12345", "u-1", "Login Failure", "Unable to log in")

    first = await create_ticket(command, repo)
    second = await create_ticket(command, repo)

    assert first == second == "T-001"
    assert len(repo.items) == 1

Demonstrating automated tests that verify "identical tool invocations do not produce duplicate writes" is far more convincing in interviews than showing success screenshots.

8. Chapter Exercises

Exercise A: Concurrency Gateway

Implement a batch model invocation function:

• Concurrency capped at 5 simultaneous requests;
• 10-second timeout per request;
• 429 errors retried up to 3 times;
• Return successful results and failure reasons without canceling the entire batch upon single failures.

Acceptance: Use a Fake Client to simulate latency, 429s, and permanent failures; write 5 test cases.

Exercise B: Asynchronous Task API

Implement /runs, /runs/{id}, and /runs/{id}/cancel three endpoints. State machine must include at least: queued, running, waiting_approval, succeeded, failed, cancelled.

Acceptance: Invalid state transitions must be rejected, such as succeeded -> running.

Exercise C: Reliable Write Tool

Implement a "Create Ticket" tool:

• Pydantic parameter validation;
• Idempotency key;
• Database unique constraint design specifications;
• Permission verification;
• Audit logging;
• Safe retry capability after timeouts.

9. High-Frequency Interview Q&A

🔥 P0: What happens when calling synchronous blocking functions inside async def?

It blocks the event loop, preventing all other coroutines on the same Worker from progressing. Solutions include using async clients, temporarily offloading to thread pools, or moving heavy CPU/blocking tasks to standalone processes and task queues. Also configure timeouts and concurrency limits.

🔥 P0: How do you design retries for Model APIs?

Classify errors first; retry only transient and safe errors; apply exponential backoff with random jitter; set maximum retry counts and global deadline budgets; enforce idempotency for write operations; trip circuit breakers or fallback to backup models upon persistent failures.

🔥 P0: Why should HTTP requests avoid waiting synchronously for Agent completion?

Long-running tasks easily exceed gateway timeouts, client retries cause duplicate executions, and server connections remain blocked for long periods. Return 202 + run_id, execute in the background, and update state via SSE/WebSocket/polling.

⭐ P1: Where should concurrency limits be enforced?

At least three layers: ingress rate limiting protects the overall service; model/tool-level Semaphore protects individual downstream systems; queue Worker concurrency protects backend resources. Ingress limiting alone cannot prevent excessive internal fan-out per request.

⭐ P1: How do you prevent duplicate creation of business data?

Use business idempotency keys, database unique constraints, transactions, and explicit state machines. When external systems do not support idempotency, maintain request-to-external-result mappings or implement compensating actions, as prompts alone cannot guarantee safety.

10. Chapter Completion Criteria

• Able to explain the difference between I/O concurrency and CPU parallelism;
• Able to write async invocations with timeouts, retries, backoff, and concurrency control;
• Able to design asynchronous task state machines;
• Able to implement and test idempotent write tools;
• Able to explain when to use thread pools, processes, queues, or dedicated GPU services.