Building Micro Agents as Production-Grade Microservices

Build production-grade AI agent systems using microservices. Covers FastAPI, gRPC, Kafka, Kubernetes, OpenTelemetry, and fault-tolerant orchestration patterns in Python. Table of Contents Introduction & Motivation Core Architecture Principles Agent Service Design The AgentRunner Loop Inter-Agent Communication Tool Registry Service Memory Architecture Context Window Management Orchestrator & Supervisor Pattern Security & Authorization Observability: Traces, Logs, Metrics Deployment on Kubernetes Scaling Strategies Fault Tolerance & Retry Strategies Testing Agent Microservices CI/CD Pipeline for Agent Services Cost Management & Token Budgeting Production Readiness Checklist Reference Architecture Diagram Introduction & Motivation Why monolithic agent systems fail in production A single-process agent that handles reasoning, tool calls, memory retrieval, and output generation works well in prototypes. In production it breaks in predictable ways: Latency coupling — one slow tool call blocks the entire inference loop Unscalable compute — you cannot scale the summarization workload independently from the search workload Blast radius — a single LLM API timeout or memory corruption takes the whole system down Zero deployment granularity — updating one tool integration requires redeploying everything No isolation for billing — impossible to attribute compute cost to individual agent functions The microservice solution Each autonomous capability becomes an independently deployable, independently scalable service with: Its own API surface (HTTP/gRPC) Its own health checks and readiness probes Its own memory scope (no shared in-process state) Its own tool bindings (resolved at runtime from a Tool Registry) Its own observability (distributed traces, metrics, structured logs) What is a Micro Agent? A micro agent is a bounded autonomous service that: Accepts a task (prompt + context + session ID) via an API call Runs a plan → act → observe loop using an LLM backend Invokes tools via a centralized Tool Registry Stores and retrieves conversation state from an external memory store Returns a typed result or emits an event to downstream consumers Key insight: A micro agent is not a “smart function” — it is a service with its own API contract, memory scope, failure modes, and SLA. Design it accordingly. Core Architecture Principles Single Responsibility Each agent owns exactly one reasoning domain. Examples: Stateless Reasoning, Stateful Memory The LLM inference step must be stateless . Memory lives in external stores: No conversation history should ever live in in-process RAM between requests. Schema-First Tool Contracts Every tool must have a JSON Schema definition published to a shared Tool Registry before any agent can invoke it. No ad-hoc function signatures. This enables: Runtime input validation before LLM output reaches backend services Auto-generated documentation Tool versioning with backwards compatibility checks Idempotent Actions Any tool call that modifies external state (send email, write to DB, trigger webhook) must be idempotent. Strategies: Use idempotency keys at the HTTP layer (pass Idempotency-Key header) Use message deduplication at the queue level (Kafka exactly-once semantics) Design tool handlers to be safe to retry: check-then-act patterns Async by Default Long-running agent tasks (multi-step research, code generation + execution) must use async task queues — not synchronous HTTP with long timeouts. Client ──► POST /tasks ──► Kafka/BullMQ ──► AgentWorker Client ──► GET /tasks/{id} ──► Redis (status polling) ◄── WebSocket/SSE push (optional) Explicit Context Boundaries Each agent invocation carries a bounded context packet — never grow unbounded message histories. A ContextManager service compresses/summarizes history before injection. Agent Service Design Project Layout Each agent is a containerized FastAPI or gRPC service with this canonical structure: agent-search/ ├── agent/ │ ├── core.py # AgentRunner: plan → act → observe loop │ ├── prompts.py # System prompt + few-shot templates │ ├── memory.py # ContextManager: load/compress/save │ ├── tools.py # Tool bindings (calls Tool Registry) │ └── schemas.py # Pydantic models for all I/O ├── api/ │ ├── routes.py # POST /run, GET /status/{task_id} │ ├── middleware.py # Auth, rate limiting, request tracing │ └── deps.py # Dependency injection: DB, Redis, LLM client ├── tests/ │ ├── unit/ │ ├── integration/ │ └── fixtures/ ├── Dockerfile ├── pyproject.toml └── k8s/ ├── deployment.yaml ├── service.yaml ├── hpa.yaml └── configmap.yaml API Contract Every agent exposes these HTTP endpoints at minimum: POST /run Submit a task (sync, short tasks only) POST /tasks Submit a task (async, returns task_id) GET /tasks/{task_id} Poll task status and result GET /health Liveness probe GET /ready Readiness probe (checks LLM + memory store) GET /metrics Prometheus metrics endpoint # agent/schemas.py from pydantic import BaseModel , Field from typing import Optional , Dict , Any from enum import Enum class TaskStatus ( str , Enum ): PENDING = " pending " RUNNING = " running " COMPLETED = " completed " FAILED = " failed " CANCELLED = " cancelled " class AgentTask ( BaseModel ): id : str session_id : str prompt : str metadata : Dict [ str , Any ] = Field ( default_factory = dict ) max_steps : int = Field ( default = 10 , ge = 1 , le = 25 ) token_budget : int = Field ( default = 8192 , ge = 512 , le = 32768 ) class AgentResult ( BaseModel ): task_id : str status : TaskStatus output : Optional [ str ] = None steps_used : int = 0 tokens_used : int = 0 tool_calls : int = 0 error : Optional [ str ] = None duration_ms : int = 0 The AgentRunner Loop Full Implementation # agent/core.py import asyncio import time from opentelemetry import trace from tenacity import retry , stop_after_attempt , wait_exponential_jitter tracer = trace . get_tracer ( name ) MAX_STEPS = 15 class AgentRunner : def init ( self , agent_id : str , config : AgentConfig ): self . agent_id = agent_id self . llm = LLMClient ( model = config . model , timeout = 30 ) self . memory = ContextManager ( agent_id , max_tokens = config . context_limit ) self . tools = ToolRegistryClient ( config . tool_registry_url ) self . metrics = AgentMetrics ( agent_id ) async def run ( self , task : AgentTask ) -> AgentResult : start = time . monotonic () with tracer . start_as_current_span ( " agent.run " ) as span : span . set_attribute ( " agent.id " , self . agent_id ) span . set_attribute ( " agent.task_id " , task . id ) span . set_attribute ( " agent.session " , task . session_id ) try : result = await self . _run_loop ( task , span ) except TokenBudgetExceeded as e : result = AgentResult ( task_id = task . id , status = TaskStatus . COMPLETED , output = e . partial_output , error = " token_budget_exceeded " ) except Exception as e : span . record_exception ( e ) result = AgentResult ( task_id = task . id , status = TaskStatus . FAILED , error = str ( e ) ) finally : result . duration_ms = int (( time . monotonic () - start ) * 1000 ) self . metrics . record ( result ) return result async def _run_loop ( self , task : AgentTask , span ) -> AgentResult : # Load available tools from registry tool_schemas = await self . tools . fetch ( agent_id = self . agent_id ) # Load and compress conversation history context = await self . memory . load ( task . session_id ) messages = build_messages ( context , task . prompt ) total_tokens = 0 tool_call_count = 0 for step in range ( task . max_steps ): span . set_attribute ( " agent.current_step " , step ) with tracer . start_as_current_span ( " agent.llm_call " ) as llm_span : response = await self . _complete_with_retry ( messages , tool_schemas ) llm_span . set_attribute ( " llm.prompt_tokens " , response . usage . prompt_tokens ) llm_span . set_attribute ( " llm.completion_tokens " , response . usage . completion_tokens ) total_tokens += response . usage . total_tokens if total_tokens > task . token_budget : raise TokenBudgetExceeded ( partial_output = response . content , tokens_used = total_tokens ) if response . finish_reason == " stop " : await self . memory . save ( task . session_id , messages + [ response . message ]) return AgentResult ( task_id = task . id , status = TaskStatus . COMPLETED , output = response . content , steps_used = step + 1 , tokens_used = total_tokens , tool_calls = tool_call_count ) if response . tool_calls : tool_call_count += len ( response . tool_calls ) results = await self . _execute_tools ( response . tool_calls ) messages . append ( response . message ) messages . extend ( tool_result_messages ( results )) # Hit max steps — return best available output return AgentResult ( task_id = task . id , status = TaskStatus . COMPLETED , output = response . content , steps_used = task . max_steps , tokens_used = total_tokens , error = " max_steps_reached " ) @retry ( stop = stop_after_attempt ( 3 ), wait = wait_exponential_jitter ( max = 15 )) async def _complete_with_retry ( self , messages , tools ): return await self . llm . complete ( messages = messages , tools = tools ) async def execute_tools ( self , tool_calls ): tasks = [ self . tools . invoke ( tc ) for tc in tool_calls ] return await asyncio . gather ( * tasks , return_exceptions = True ) Inter-Agent Communication Pattern Selection Matrix gRPC Service Definition For synchronous sub-agent calls, gRPC provides strong typing, bidirectional streaming, and efficient binary serialization. // proto/agent_service.proto syntax = "proto3" ; package agents . v1 ; service AgentService { rpc RunTask ( TaskRequest ) returns ( TaskResponse ); rpc StreamSteps ( TaskRequest ) returns ( stream StepEvent ); rpc Health ( HealthRequest ) returns ( HealthResponse ); } message TaskRequest { string task_id = 1 ; string session_id = 2 ; string prompt = 3 ; map < string , string > metadata = 4 ; int32 max_steps = 5 ; int32 token_budget = 6 ; } message TaskResponse { string task_id = 1 ; string status = 2 ; string output = 3 ; int32 steps_used = 4 ; int32 tokens_used = 5 ; string error = 6 ; } message StepEvent { int32 step_number = 1 ; string type = 2 ; // "llm_call" | "tool_call" | "tool_result" string content = 3 ; } Kafka Event Schema For async pipeline handoffs between agents, use Avro or JSON schemas registered in a Schema Registry. { "schema" : { "type" : "record" , "name" : "AgentTaskEvent" , "namespace" : "com.myco.agents.v1" , "fields" : [ { "name" : "task_id" , "type" : "string" }, { "name" : "source_agent" , "type" : "string" }, { "name" : "target_agent" , "type" : "string" }, { "name" : "session_id" , "type" : "string" }, { "name" : "prompt" , "type" : "string" }, { "name" : "context" , "type" : { "type" : "map" , "values" : "string" }}, { "name" : "created_at" , "type" : { "type" : "long" , "logicalType" : "timestamp-millis" }} ] } } Kafka Producer (in Orchestrator) # In orchestrator when dispatching to agent-search from aiokafka import AIOKafkaProducer import json async def dispatch_to_agent ( target_agent : str , task : AgentTask ): producer = AIOKafkaProducer ( bootstrap_servers = KAFKA_BROKERS ) await producer . start () try : event = { " task_id " : task . id , " source_agent " : " orchestrator " , " target_agent " : target_agent , " session_id " : task . session_id , " prompt " : task . prompt , " created_at " : int ( time . time () * 1000 ) } await producer . send_and_wait ( topic = f " agent.tasks. { target_agent } " , value = json . dumps ( event ). encode (), key = task . session_id . encode (), # partition by session headers = [( " trace-id " , get_current_trace_id (). encode ())] ) finally : await producer . stop () Tool Registry Service Architecture The Tool Registry is a centralized FastAPI service that stores, validates, and serves tool definitions. It acts as a typed API gateway for all agent→tool traffic. Tool Registration Schema # Tool self-registers on startup class ToolDefinition ( BaseModel ): name : str version : str description : str parameters : Dict [ str , Any ] # JSON Schema returns : Dict [ str , Any ] # JSON Schema endpoint : str # where registry routes calls health_url : str auth_type : str # "api_key" | "oauth2" | "none" rate_limit : int # calls per minute per agent timeout_ms : int = 10000 # Registration call at tool service startup @app.on_event ( " startup " ) async def register_tool (): registry = ToolRegistryClient ( TOOL_REGISTRY_URL ) await registry . register ( ToolDefinition ( name = " web_search " , version = " 2.1.0 " , description = " Search the web and return ranked results " , parameters = { " type " : " object " , " properties " : { " query " : { " type " : " string " , " maxLength " : 500 }, " num_results " : { " type " : " integer " , " minimum " : 1 , " maximum " : 20 } }, " required " : [ " query " ] }, returns = { " type " : " array " , " items " : { " type " : " object " , " properties " : { " url " : { " type " : " string " }, " title " : { " type " : " string " }, " snippet " : { " type " : " string " } } } }, endpoint = f " { SERVICE_URL } /invoke " , health_url = f " { SERVICE_URL } /health " , auth_type = " api_key " , rate_limit = 60 , timeout_ms = 8000 )) Registry Validation Layer # Tool Registry validates before forwarding async def invoke_tool ( agent_id : str , tool_name : str , params : dict ): tool = await db . get_tool ( tool_name ) if not tool : raise ToolNotFoundError ( tool_name ) # Validate against JSON Schema jsonschema . validate ( params , tool . parameters ) # raises on invalid input # Check rate limit if not await rate_limiter . check ( agent_id , tool_name , tool . rate_limit ): raise RateLimitExceeded ( f " { tool_name } limit: { tool . rate_limit } /min " ) # Forward to tool service with timeout async with httpx . AsyncClient ( timeout = tool . timeout_ms / 1000 ) as client : response = await client . post ( tool . endpoint , json = { " params " : params }, headers = { " X-Agent-Id " : agent_id , " X-Request-Id " : str ( uuid4 ())} ) response . raise_for_status () return response . json () Memory Architecture Memory Tier Selection ContextManager Implementation # agent/memory.py import json from redis.asyncio import Redis from qdrant_client import QdrantClient from typing import List class ContextManager : def init ( self , agent_id : str , max_tokens : int = 4096 ): self . agent_id = agent_id self . max_tokens = max_tokens self . redis = Redis . from_url ( REDIS_URL ) self . qdrant = QdrantClient ( QDRANT_URL ) self . embedder = EmbeddingClient () async def load ( self , session_id : str ) -> List [ dict ]: # 1. Load recent turns from Redis raw = await self . redis . get ( f " session: { session_id } :messages " ) messages = json . loads ( raw ) if raw else [] # 2. Retrieve semantically relevant past context if messages : last_user_msg = next ( m for m in reversed ( messages ) if m [ " role " ] == " user " ) embedding = await self . embedder . embed ( last_user_msg [ " content " ]) relevant = await self . qdrant . search ( collection_name = f " agent { self . agent_id } _memory " , query_vector = embedding , limit = 3 ) # Prepend as system context for hit in relevant : messages . insert ( 0 , { " role " : " system " , " content " : f " [Past context] { hit . payload [ ' summary ' ] } " }) # 3. Compress if over token limit return await self . _compress_if_needed ( messages ) async def save ( self , session_id : str , messages : List [ dict ]): # Save last 20 turns to Redis recent = messages [ - 20 :] await self . redis . setex ( f " session: { session_id } :messages " , 86400 , # 24h TTL json . dumps ( recent ) ) # If session is long, generate and store a summary in vector DB if len ( messages ) > 30 : summary = await self . summarize ( messages ) embedding = await self . embedder . embed ( summary ) await self . qdrant . upsert ( collection_name = f " agent { self . agent_id } _memory " , points = [{ " id " : session_id , " vector " : embedding , " payload " : { " summary " : summary , " session_id " : session_id } }] ) async def _compress_if_needed ( self , messages : List [ dict ]) -> List [ dict ]: token_count = estimate_tokens ( messages ) if token_count <= self . max_tokens : return messages # Keep system messages + last N user/assistant turns system_msgs = [ m for m in messages if m [ " role " ] == " system " ] recent_turns = messages [ - 12 :] # last 6 exchanges return system_msgs + recent_turns Context Window Management Token Estimation import tiktoken def estimate_tokens ( messages : list , model : str = " gpt-4o " ) -> int : enc = tiktoken . encoding_for_model ( model ) total = 0 for msg in messages : total += 4 # per-message overhead total += len ( enc . encode ( msg . get ( " content " , "" ) or "" )) if " tool_calls " in msg : for tc in msg [ " tool_calls " ]: total += len ( enc . encode ( json . dumps ( tc ))) return total class TokenBudget : def init ( self , total : int , model : str ): self . total = total self . model = model self . used = 0 self . reserved = 1024 # always reserve for output @property def available_for_input ( self ): return self . total - self . reserved - self . used def consume ( self , tokens : int ): self . used += tokens if self . used > self . total - self . reserved : raise TokenBudgetExceeded ( tokens_used = self . used ) Orchestrator & Supervisor Pattern Orchestrator: Task Decomposition The Orchestrator is itself an agent microservice, but its role is planning and coordination rather than execution. # orchestrator/core.py class OrchestratorAgent : async def execute ( self , user_request : str , session_id : str ) -> str : # Step 1: Decompose into a DAG of sub-tasks plan = await self . planner . decompose ( user_request ) # Returns: [{"id": "t1", "agent": "search", "task": "...", "deps": []}, # {"id": "t2", "agent": "summarize", "task": "...", "deps": ["t1"]}, # {"id": "t3", "agent": "email", "task": "...", "deps": ["t2"]}] # Step 2: Execute in topological order, parallel where possible results = {} for wave in topological_waves ( plan ): # All tasks in a wave have their deps satisfied wave_results = await asyncio . gather ( * [ self . supervisor . dispatch ( step , results ) for step in wave ]) for step , result in zip ( wave , wave_results ): results [ step [ " id " ]] = result # Step 3: Synthesize final output return await self . synthesizer . merge ( results , user_request ) def topological_waves ( plan : list ) -> list : """ Return plan steps grouped into parallel execution waves. """ completed = set () waves = [] remaining = list ( plan ) while remaining : wave = [ s for s in remaining if all ( d in completed for d in s [ " deps " ])] waves . append ( wave ) completed . update ( s [ " id " ] for s in wave ) remaining = [ s for s in remaining if s [ " id " ] not in completed ] return waves Supervisor: Retry & Escalation class Supervisor : def init ( self , agent_clients : dict ): self . agent_clients = agent_clients async def dispatch ( self , step : dict , context : dict ) -> StepResult : task_prompt = self . _inject_context ( step [ " task " ], context , step [ " deps " ]) for attempt in range ( 3 ): try : return await asyncio . wait_for ( self . agent_clients [ step [ " agent " ]]. run ( task_prompt ), timeout = 60.0 ) except asyncio . TimeoutError : if attempt == 2 : raise SupervisorEscalation ( step , " timeout_after_3_attempts " ) await asyncio . sleep ( 2 ** attempt ) # 1s, 2s, 4s except AgentError as e : if e . is_unrecoverable : raise SupervisorEscalation ( step , str ( e )) await asyncio . sleep ( 2 ** attempt ) def _inject_context ( self , task : str , results : dict , dep_ids : list ) -> str : context_parts = [ results [ dep_id ]. output for dep_id in dep_ids if dep_id in results ] if context_parts : return f " Context from previous steps: \n { chr ( 10 ). join ( context_parts ) } \n\n Task: { task } " return task Security & Authorization Agent Identity & JWT Verification Each agent service must verify that incoming requests are from authorized callers. Use sh