Agent SDK engineer coding interview questions

As asked

Write a TypeScript tool definition for creating a support ticket, then explain how you would make the schema safe for agent use.

Sample answer outline

The schema should make required fields explicit, restrict enums, and avoid accepting vague free-form blobs for important state. Tool descriptions should say when to call the tool and when not to call it, but the server must still validate permissions and business rules. A good answer includes idempotency keys or deduplication because agents may retry after timeouts. The tool should return a small structured result the agent can reason over, not an entire database record. Candidates often make tools too broad, which increases accidental side effects and makes eval failures hard to diagnose.

Reference implementation (typescript)

import { z } from "zod";

export const createSupportTicketTool = {
  name: "create_support_ticket",
  description:
    "Create one support ticket after the user has described a concrete problem. Do not call for general questions.",
  inputSchema: z.object({
    customerId: z.string().min(1),
    idempotencyKey: z.string().uuid(),
    severity: z.enum(["low", "normal", "high", "urgent"]),
    subject: z.string().min(5).max(120),
    description: z.string().min(20).max(4000),
    orderId: z.string().optional(),
  }),
  async execute(input: unknown) {
    const args = this.inputSchema.parse(input);
    return {
      ticketId: await createTicket(args),
      status: "created" as const,
    };
  },
};

Expect these follow-ups

• How do you version this tool without breaking old agents?
• Where does human approval fit?
• What should the tool return on validation failure?

As asked

Write a TypeScript class that acts as a tool registry. It should support registering a tool with a name, a JSON Schema for its arguments, and a handler function. It should expose a method that, given a tool name and raw arguments object, validates the arguments against the schema and calls the handler, throwing a descriptive error on validation failure.

Sample answer outline

A strong solution defines a generic Tool<T> type with name, schema, and handler fields, uses Ajv to compile and cache the validator at registration time, exposes an async dispatch(name, args) method that looks up the tool, runs the validator, throws ToolValidationError with the ajv errors on failure, and awaits the handler. Bonus: returning a typed result or using zod for schema definition.

Reference implementation (typescript)

import Ajv from 'ajv';

interface Tool<T = unknown> {
  name: string;
  schema: object;
  handler: (args: T) => Promise<unknown>;
}

class ToolRegistry {
  private tools = new Map<string, Tool>();
  private ajv = new Ajv();

  register<T>(tool: Tool<T>): void {
    // TODO: compile schema and store validator
  }

  async dispatch(name: string, args: unknown): Promise<unknown> {
    // TODO: validate args, call handler
  }
}

Expect these follow-ups

• How would you extend this registry to support streaming handlers that yield partial results?
• How would you serialize this registry to JSON so a remote LLM client can consume the tool list?

As asked

Write a Python async function run_agent(messages, tools, client) that runs an agent loop: call the LLM, check if it wants to call tools, execute them concurrently, append results, and repeat until the model stops calling tools or max_steps is reached. Return the final assistant message.

Sample answer outline

A strong solution uses asyncio.gather to run tool calls in parallel, correctly constructs the tool result messages with matching tool_call_ids, guards against infinite loops with a step counter, handles the case where the model emits zero tool calls (done), and propagates tool exceptions as error tool results rather than crashing. Candidates should also handle the provider-specific message format.

Reference implementation (python)

import asyncio
from typing import Any

async def run_agent(
    messages: list[dict],
    tools: dict[str, Any],
    client,
    max_steps: int = 20,
) -> dict:
    for step in range(max_steps):
        response = await client.messages.create(
            model='claude-opus-4-5',
            messages=messages,
            tools=list(tools.values()),
        )
        # TODO: check stop_reason, dispatch tool calls, append results
    raise RuntimeError('max_steps exceeded')

Expect these follow-ups

• How would you add streaming support so the user sees tokens as they are generated?
• How would you make this function resumable after a crash?

As asked

Write a TypeScript generic utility withRetry(fn, options) that wraps an async function with exponential backoff, optional jitter, a maximum retry count, and a predicate to decide whether an error is retriable. It should surface a final error if all retries are exhausted.

Sample answer outline

A strong solution computes delay as baseMs * 2^attempt plus a random jitter between 0 and jitterMs, uses a type-safe retriable predicate that defaults to always-true, throws a RetryExhaustedError wrapping the last underlying error, and does not swallow non-retriable errors on the first attempt. Candidates should avoid using a fixed sleep loop and instead use a single awaitable delay helper.

Reference implementation (typescript)

interface RetryOptions {
  maxAttempts: number;
  baseMs: number;
  jitterMs?: number;
  isRetriable?: (err: unknown) => boolean;
}

async function withRetry<T>(
  fn: () => Promise<T>,
  opts: RetryOptions,
): Promise<T> {
  // TODO: implement
}

Expect these follow-ups

• How would you add a total timeout across all attempts rather than a per-attempt timeout?
• How would you make this work with AbortSignal so the caller can cancel mid-retry?

As asked

What makes a sample app credible enough that developers copy from it without carrying bad patterns into production?

Sample answer outline

A credible sample app is small, runnable, and honest about what is production-grade and what is simplified. It should use current SDK APIs, secure secret handling, typed configuration, meaningful error handling, and tests around the integration path. The candidate should avoid building a toy that hides authentication, pagination, retries, or deployment concerns if those are central to real adoption. Strong answers include maintenance: version pinning, dependency updates, CI, and ownership when the API changes. Interviewers look for judgement about where completeness helps learning and where it becomes distracting.

Expect these follow-ups

• What is the first thing you would remove from an overbuilt sample app?
• How do you stop copied sample code from leaking secrets?
• How do you keep examples current across SDK versions?

As asked

Write a Python async function that accepts a natural language query, embeds it using OpenAI's text-embedding-3-small, queries a pgvector table for the top 5 nearest neighbors, and returns the results as a list of dicts with id, content, and score. Use asyncpg for the database connection.

Sample answer outline

A strong solution uses the embeddings API correctly (input is a list of strings), converts the returned embedding to a list for formatting into the pgvector cast, uses the <=> cosine distance operator in the ORDER BY clause, limits to 5 rows, and maps the results to the expected dict schema. Candidates should parameterize the embedding to avoid SQL injection and handle the asyncpg connection as a passed-in pool rather than opening a new connection per call.

Reference implementation (python)

import asyncpg
from openai import AsyncOpenAI

oai = AsyncOpenAI()

async def retrieve(
    query: str,
    pool: asyncpg.Pool,
    top_k: int = 5,
) -> list[dict]:
    # 1. embed the query
    # 2. query pgvector for nearest neighbors
    # 3. return list of {id, content, score}
    ...

Expect these follow-ups

• How would you add metadata filters so the caller can restrict results to a specific document category?
• What index type would you create on the vector column and why?

As asked

Write a TypeScript function compressHistory(messages, maxTokens, estimateTokens, summarize) that keeps the most recent messages up to maxTokens tokens, summarizes older messages into a single synthetic assistant message, and returns the resulting compressed messages array. The system message must always be preserved at position 0.

Sample answer outline

A strong solution scans from the most recent message backward, accumulates token counts, stops when adding the next older message would exceed the budget, calls summarize on the skipped slice, injects the summary as a synthetic message right after the system message, and preserves message ordering. Edge cases include the system message alone exceeding the budget and an empty history.

Reference implementation (typescript)

type Message = { role: string; content: string };

async function compressHistory(
  messages: Message[],
  maxTokens: number,
  estimateTokens: (msgs: Message[]) => number,
  summarize: (msgs: Message[]) => Promise<string>,
): Promise<Message[]> {
  // system message is messages[0]
  // TODO: implement sliding window compression
}

Expect these follow-ups

• How would you test that the compressed history still allows the agent to answer questions that depend on the summarized content?
• How would you handle tool messages and tool result pairs that must stay together to remain valid?

As asked

Write a TypeScript class SlidingWindowRateLimiter that limits calls to N per second using a sliding window algorithm. It should expose an async acquire() method that resolves immediately if under the limit and waits (with a timeout option) if the limit is reached.

Sample answer outline

A strong solution tracks timestamps of recent calls in a circular buffer or sorted array, evicts entries older than the window on each acquire call, blocks the caller by awaiting a promise that resolves when a slot opens, and rejects with a RateLimitTimeoutError if the timeout expires. Candidates should avoid busy-waiting and instead schedule the next available slot precisely.

Reference implementation (typescript)

class SlidingWindowRateLimiter {
  private timestamps: number[] = [];

  constructor(
    private readonly limit: number,
    private readonly windowMs: number,
  ) {}

  async acquire(timeoutMs = 5000): Promise<void> {
    // TODO: implement sliding window logic
  }
}

Expect these follow-ups

• How would you adapt this to a token bucket model where different calls consume different amounts?
• How would you share this rate limiter across multiple Node.js worker threads?

As asked

Write a Python class RedisCheckpointStore that saves and loads agent run state (a list of messages) to Redis, keyed by run_id and step_number. It should support listing all checkpoints for a run and deleting a completed run's checkpoints to avoid unbounded storage growth.

Sample answer outline

A strong solution serializes messages with json.dumps, uses a Redis hash or sorted set keyed by run_id with step_number as the field or score, sets a TTL on keys for completed runs, uses a pipeline to batch writes for efficiency, and exposes list_steps(run_id) and delete_run(run_id) methods. Candidates should handle Redis connection errors gracefully and discuss the tradeoff between storing raw JSON versus compressed bytes.

Reference implementation (python)

import redis.asyncio as aioredis
import json
from typing import Any

class RedisCheckpointStore:
    def __init__(self, client: aioredis.Redis, ttl_seconds: int = 86400):
        self.client = client
        self.ttl = ttl_seconds

    async def save(self, run_id: str, step: int, messages: list[dict]) -> None:
        ...

    async def load(self, run_id: str, step: int) -> list[dict]:
        ...

    async def list_steps(self, run_id: str) -> list[int]:
        ...

    async def delete_run(self, run_id: str) -> None:
        ...

Expect these follow-ups

• How would you handle a race condition where two workers try to write checkpoint for the same step?
• How would you migrate checkpoints when your message schema changes?

As asked

Write a TypeScript async generator function parseSSE(response: Response) that reads a Fetch API response body as a stream of Server-Sent Events, parses each data line as JSON, and yields the parsed objects. It should handle multi-line data fields and skip comment lines.

Sample answer outline

A strong solution uses response.body.getReader() with a TextDecoder, buffers partial chunks across read calls, splits on double newline event boundaries, extracts data: prefix lines and concatenates multi-line data, skips lines starting with colon (comments), and closes the reader cleanly on done or error. Candidates should handle the done: [DONE] sentinel and avoid loading the entire stream into memory.

Reference implementation (typescript)

async function* parseSSE(
  response: Response,
): AsyncGenerator<unknown> {
  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = '';
  // TODO: read chunks, parse SSE events, yield JSON objects
}

Expect these follow-ups

• How would you add support for retry: fields that tell you to reconnect after a given delay?
• How would you unit-test this generator with a mocked response stream?

As asked

Some tools are deterministic given the same arguments (like a currency conversion API). Write a Python class ToolResultCache with a fixed capacity that caches tool results by (tool_name, frozenset of args), evicts the least recently used entry when full, and supports async get and set.

Sample answer outline

A strong solution uses an OrderedDict to maintain insertion and access order, moves a key to the end on access (most recently used), removes the first key (least recently used) on overflow, serializes the args dict to a canonical frozenset of items for hashing, and wraps operations in an asyncio.Lock since the cache is shared state. Candidates should discuss cache invalidation strategies.

Reference implementation (python)

import asyncio
from collections import OrderedDict
from typing import Any

class ToolResultCache:
    def __init__(self, capacity: int):
        self.capacity = capacity
        self._cache: OrderedDict[tuple, Any] = OrderedDict()
        self._lock = asyncio.Lock()

    def _key(self, tool_name: str, args: dict) -> tuple:
        return (tool_name, frozenset(args.items()))

    async def get(self, tool_name: str, args: dict) -> Any | None:
        ...

    async def set(self, tool_name: str, args: dict, result: Any) -> None:
        ...

Expect these follow-ups

• How would you add a TTL per entry so stale results expire automatically?
• How would you make this cache survive a process restart by persisting to Redis?

As asked

Implement a min-heap based priority queue in Python that schedules agent tasks by priority (lower number = higher priority), supports adding tasks, popping the highest-priority task, and peeking without removal. Make it thread-safe for use from multiple asyncio tasks.

Sample answer outline

A strong solution uses heapq internally with a (priority, insertion_order, task) tuple to break ties consistently, wraps mutating operations in an asyncio.Lock, exposes async push and async pop, raises QueueEmptyError on pop from an empty queue, and uses insertion_order as a tiebreaker to give FIFO behavior within the same priority. Candidates should avoid using a threading.Lock in async code.

Reference implementation (python)

import heapq
import asyncio
from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class PrioritizedTask:
    priority: int
    index: int
    task: Any = field(compare=False)

class AgentTaskQueue:
    def __init__(self):
        self._heap: list[PrioritizedTask] = []
        self._counter = 0
        self._lock = asyncio.Lock()

    async def push(self, task: Any, priority: int) -> None:
        ...

    async def pop(self) -> Any:
        ...

Expect these follow-ups

• How would you add the ability to cancel a task that is already in the queue?
• How would you adapt this for a multi-process setup where workers run in separate OS processes?

As asked

Given a dictionary mapping task names to lists of their dependencies, write a Python function that returns a valid execution order (topological sort) and raises a CyclicDependencyError if a cycle is detected. This is used to determine which subtasks can run in parallel.

Sample answer outline

A strong solution uses Kahn's algorithm or DFS with color marking (white/gray/black), correctly raises CyclicDependencyError with the cycle nodes included in the message, and handles nodes with no dependencies as valid roots. Candidates should also identify which tasks can run in parallel by grouping tasks with no remaining dependencies at each iteration of Kahn's algorithm.

Reference implementation (python)

from typing import Dict, List

def topological_sort(
    graph: Dict[str, List[str]]
) -> List[str]:
    """
    graph[task] = list of tasks that must complete before task
    Returns execution order or raises CyclicDependencyError.
    """
    ...

class CyclicDependencyError(Exception):
    pass

Expect these follow-ups

• How would you modify this to return batches of tasks that can run concurrently rather than a flat ordered list?
• How would you handle a task graph that has hundreds of nodes efficiently?