The translation agent — a small program that converts text between languages — printed “Translating from Spanish to English,” then sat there. Not crashed. Just waiting.
Meanwhile the fact-checking agent had already finished its search three seconds ago. It had no way to tell the coordinator it was done. The coordinator was locked on the translator’s stream, checking over and over for a completion signal that wasn’t coming. The translator was waiting on a language-detection call to an external service, and that call had silently timed out.
The user saw a spinner.
That’s when I realized my modular agent system had the right architecture and the wrong communication model. Each skill worked fine in isolation. Wire them together, and the whole thing was one slow component away from looking broken.
I rewired it with Server-Sent Events — a way for the server to push updates to the client over a plain HTTP connection — and the whole system went from fragile to boring. Boring is what you want in infrastructure.
The important parts were the architecture, the wrong turns, and why SSE fit this job better than the alternatives.
The monolith ran out of room
Before modular, I had the monolith. One agent, one codebase, every capability wired into the same process. It worked for the first two skills. By the fifth, it was a mess.
Three problems kept showing up. Adding translation broke search when its newer language-model library shifted a dependency the search feature used. I spent an afternoon untangling that tight coupling — two pieces that cannot change independently without surprises.
A one-line grounding fix also meant restarting the full agent, even if a user was mid-request. And translation wanted the latest language-model client while grounding needed a search API client. The monolith forced them into one environment and one set of versions. That is technology lock-in: the system’s shape decides which tools fit inside it.
As the number of capabilities grew past three or four, each new addition felt like surgery on a patient that was still awake. I needed a different approach.
Three pieces instead of one
The alternative has three pieces. The agent core handles routing, session state, and the public API; it does not do the work. Skills are standalone services for translation, grounding, or search. Each owns its dependencies, can deploy independently, and exposes one endpoint. The communication layer is how the core calls a skill and streams results back to the client. SSE lives there.
Here’s the core in practice:
class AgentCore:
def __init__(self):
self.skills = {} # Registered skills
self.state = {} # Agent state
self.session_store = SessionStore() # For persisting sessions
def register_skill(self, skill_name, skill_endpoint):
"""Register a skill with the agent core."""
self.skills[skill_name] = skill_endpoint
logger.info(f"Registered skill: {skill_name} at {skill_endpoint}")
async def execute_skill(self, skill_name, input_data):
"""Execute a specific skill and return results."""
if skill_name not in self.skills:
raise SkillNotFoundError(f"Skill '{skill_name}' not registered")
skill_endpoint = self.skills[skill_name]
# Create SSE client to receive real-time updates
async with SSEClient(f"{skill_endpoint}/execute") as client:
# Send initial request
await client.send_request(input_data)
# Process SSE events
async for event in client:
event_data = json.loads(event.data)
event_type = event.event or "update"
# Handle different event types
if event_type == "status":
self.state["status"] = event_data["status"]
yield {"type": "status", "data": event_data}
elif event_type == "thinking":
yield {"type": "thinking", "data": event_data["content"]}
elif event_type == "result":
self.state["last_result"] = event_data
yield {"type": "result", "data": event_data}
elif event_type == "error":
yield {"type": "error", "data": event_data["message"]}
async def process_request(self, user_request):
"""Process a user request, determining which skill to use."""
# Determine appropriate skill based on request
skill_name = self.determine_skill(user_request)
# Execute the skill and stream results
async for update in self.execute_skill(skill_name, user_request):
yield update
def determine_skill(self, request):
"""Determine which skill should handle a request."""
# This could use LLM-based routing, rule-based matching, etc.
# Simplified implementation for demonstration
if "translate" in request.get("action", "").lower():
return "translation"
elif "ground" in request.get("action", "").lower():
return "grounding"
else:
return "default"The method determine_skill is the dispatcher. It reads the request and picks which skill to call. In production you’d use a language model for routing — send the request to Claude, ask “which skill handles this?” — but the pattern is the same: inspect, route, stream.
Skills as Independent Services
Each skill runs as its own process — a small web server (FastAPI, built on Python) that exposes one endpoint (/execute) and streams events back to the core as they happen.
Here’s what a grounding skill looks like — it checks statements against web search results to verify their accuracy:
class GroundingSkill:
def __init__(self):
self.llm_client = LLMClient() # LLM service client
self.search_client = SearchClient() # Web search client
async def ground(self, statement, search_results=None):
"""Ground a statement using search results or by performing a search."""
# Get search results if not provided
if not search_results:
search_results = await self.search_client.search(statement)
# Use LLM to evaluate statement against search results
evaluation = await self.llm_client.evaluate_grounding(
statement=statement,
sources=search_results
)
return {
"statement": statement,
"is_grounded": evaluation["is_grounded"],
"confidence": evaluation["confidence"],
"sources": evaluation["relevant_sources"],
"reasoning": evaluation["reasoning"]
}
async def process_grounding_request(self, request, sse_response):
"""Process a grounding request with SSE updates."""
statement = request.get("statement", "")
# Send status update
await sse_response.send(
data=json.dumps({"status": "Searching for relevant information"}),
event="status"
)
# Perform search
search_results = await self.search_client.search(statement)
# Send thinking update
await sse_response.send(
data=json.dumps({"content": "Analyzing search results for relevant information"}),
event="thinking"
)
# Ground the statement
result = await self.ground(statement, search_results)
# Send final result
await sse_response.send(
data=json.dumps(result),
event="result"
)This skill can be deployed as a separate microservice with FastAPI:
from fastapi import FastAPI, Request
from sse_starlette.sse import EventSourceResponse
app = FastAPI()
grounding_skill = GroundingSkill()
@app.post("/execute")
async def execute_skill(request: Request):
"""Execute the grounding skill with SSE updates."""
request_data = await request.json()
async def event_generator():
try:
await grounding_skill.process_grounding_request(request_data, SSEResponse())
except Exception as e:
# Send error event
await SSEResponse().send(
data=json.dumps({"message": str(e)}),
event="error"
)
return EventSourceResponse(event_generator())
class SSEResponse:
"""Helper class to send SSE events."""
async def send(self, data, event=None):
if event:
return {"event": event, "data": data}
return {"data": data}The key pattern is status → thinking → result. The skill emits each event as it works, and the core sends it to the client immediately. The user sees progress instead of a spinner.
I first had skills return one JSON blob after finishing. That caused the deadlock from the opening: the core could not show progress until the full response arrived. SSE fixed it by letting skills emit events instead of bundling everything at the end.
Why I chose SSE
This is the load-bearing decision. The core and skills need a real-time channel, and there are two obvious options: WebSockets (bidirectional, persistent connections) and Server-Sent Events (server pushes to client, one direction, over plain HTTP).
I chose SSE because it is one-way, which matches this flow: the client sends one POST and listens. You do not need a two-way socket, which means less code and fewer edge cases. It uses plain HTTP, so there is no upgrade negotiation or WebSocket-aware proxy configuration; it works through every reverse proxy, load balancer, and CDN without special configuration. Browsers reconnect automatically when the connection drops, so you do not write reconnection logic. And named events such as event: status and event: result match the updates a skill emits.
The implementation includes custom SSE handling on both client and server sides:
the mechanism — why SSE, how state flows, try it yourself give me the detail
Why SSE over WebSockets here. SSE is unidirectional (server → client), which matches the agent’s actual data flow: the client fires one POST, then listens. Because it rides plain HTTP/1.1, it works through every reverse proxy and load balancer without upgrade negotiation. The browser reconnects automatically on drop. WebSockets buy you bidirectionality you don’t need and multiplexing complexity you don’t want.
The server-side stack. Each skill is a standalone FastAPI process that returns an EventSourceResponse from sse-starlette. The agent core calls skills over plain HTTP using aiohttp, streaming the response line-by-line. State that must survive across skill calls (routing context, session metadata) lives in Redis — the SharedStateManager above uses aioredis with a TTL so stale sessions self-clean. Each skill container is isolated in its own Docker network alias, so the core discovers them purely by environment variable, and you can restart or swap a skill without touching the core.
Smoke-test your SSE endpoint in one line:
curl -N -X POST http://localhost:8081/execute \
-H "Content-Type: application/json" \
-H "Accept: text/event-stream" \
-d '{"statement": "The sky is blue"}'You should see raw event: / data: frames scroll in real time. If you get a single JSON blob instead, EventSourceResponse isn’t wrapping your generator — check that your generator yields dicts, not returns them.
The translation skill
Here’s another skill, wired the same way, but built for language translation:
class TranslationSkill:
"""Skill for translating text between languages."""
def __init__(self):
self.llm_client = LLMClient()
self.supported_languages = self._load_supported_languages()
def _load_supported_languages(self):
"""Load supported language codes and names."""
# In practice, this would load from a configuration file
return {
"en": "English",
"es": "Spanish",
"fr": "French",
"de": "German",
"zh": "Chinese",
"ja": "Japanese",
# ... more languages
}
async def translate(self, text, source_lang, target_lang):
"""Translate text from source language to target language."""
# Validate languages
if source_lang not in self.supported_languages:
raise ValueError(f"Unsupported source language: {source_lang}")
if target_lang not in self.supported_languages:
raise ValueError(f"Unsupported target language: {target_lang}")
# Use LLM for translation
prompt = f"""
Translate the following text from {self.supported_languages[source_lang]} to {self.supported_languages[target_lang]}:
{text}
Provide only the translated text without any additional explanation.
"""
response = await self.llm_client.generate(prompt)
return {
"original_text": text,
"translated_text": response.strip(),
"source_language": source_lang,
"target_language": target_lang
}
async def process_translation_request(self, request, sse_response):
"""Process a translation request with SSE updates."""
text = request.get("text", "")
source_lang = request.get("source_lang", "auto")
target_lang = request.get("target_lang", "en")
# If source language is auto, detect it
if source_lang == "auto":
await sse_response.send(
data=json.dumps({"status": "Detecting language"}),
event="status"
)
source_lang = await self._detect_language(text)
# Send status update
await sse_response.send(
data=json.dumps({
"status": f"Translating from {self.supported_languages.get(source_lang, source_lang)} to {self.supported_languages.get(target_lang, target_lang)}"
}),
event="status"
)
# Send thinking update
await sse_response.send(
data=json.dumps({
"content": "Processing translation request..."
}),
event="thinking"
)
# Perform translation
result = await self.translate(text, source_lang, target_lang)
# Send final result
await sse_response.send(
data=json.dumps(result),
event="result"
)
async def _detect_language(self, text):
"""Detect the language of the input text."""
prompt = f"""
Identify the language of the following text. Respond with only the ISO 639-1
language code (e.g., 'en' for English, 'es' for Spanish):
{text}
"""
response = await self.llm_client.generate(prompt)
detected_lang = response.strip().lower()
# Default to English if detection fails
if detected_lang not in self.supported_languages:
return "en"
return detected_langThe translation skill shows the same SSE event flow: status → thinking → result. Every skill follows this contract, which is what makes the system composable — the core doesn’t need to know how translation works, only that it’ll emit these events in this order.
Adding MCP
For consistent message formatting across skills, I integrated with MCP (Model Control Protocol — a standard way for AI models and agents to structure their messages to clients):
class MCPHandler:
"""Handles MCP protocol formatting for SSE events."""
@staticmethod
def format_thinking(thinking_text):
"""Format thinking update in MCP format."""
return {
"type": "thinking",
"content": thinking_text,
"timestamp": datetime.now().isoformat()
}
@staticmethod
def format_status(status_text):
"""Format status update in MCP format."""
return {
"type": "status",
"status": status_text,
"timestamp": datetime.now().isoformat()
}
@staticmethod
def format_result(result_data):
"""Format final result in MCP format."""
return {
"type": "result",
"content": result_data,
"timestamp": datetime.now().isoformat()
}
@staticmethod
def format_error(error_message):
"""Format error in MCP format."""
return {
"type": "error",
"message": error_message,
"timestamp": datetime.now().isoformat()
}This is thin — just a timestamp wrapper around each event type — but it means every skill speaks the same event vocabulary. A thinking event from the translation skill has the same shape as one from the grounding skill. The client-side code doesn’t care which skill produced it.
The client listens
On the frontend, the client opens the SSE stream and dispatches events by type:
// Client-side JavaScript for connecting to the agent
const connectToAgent = async (request) => {
const response = await fetch('/api/agent/process', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Accept': 'text/event-stream'
},
body: JSON.stringify(request)
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
// Process the SSE stream
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value);
const events = chunk.split('\n\n').filter(Boolean);
for (const eventText of events) {
const eventData = parseSSEEvent(eventText);
switch (eventData.type) {
case 'status':
updateStatus(eventData.data.status);
break;
case 'thinking':
updateThinking(eventData.data.content);
break;
case 'result':
displayResult(eventData.data);
break;
case 'error':
showError(eventData.data.message);
break;
}
}
}
};
const parseSSEEvent = (eventText) => {
const lines = eventText.split('\n');
const eventData = {};
for (const line of lines) {
if (line.startsWith('event: ')) {
eventData.type = line.substring(7);
} else if (line.startsWith('data: ')) {
eventData.data = JSON.parse(line.substring(6));
}
}
return eventData;
};The pattern is simple: open a POST, get back a stream of text, split it on SSE frame boundaries (\n\n), parse each frame into a type and JSON data payload, and update the UI accordingly. Nothing exotic.
Deployment
Each component gets its own container — the core and each skill are isolated processes that talk over the network:
# Example docker-compose.yml for deploying the system
version: '3'
services:
agent-core:
build: ./agent-core
ports:
- "8080:8080"
environment:
- GROUNDING_SKILL_URL=http://grounding-skill:8081
- TRANSLATION_SKILL_URL=http://translation-skill:8082
depends_on:
- grounding-skill
- translation-skill
networks:
- agent-network
grounding-skill:
build: ./grounding-skill
ports:
- "8081:8081"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
networks:
- agent-network
translation-skill:
build: ./translation-skill
ports:
- "8082:8082"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
networks:
- agent-network
networks:
agent-network:
driver: bridgeFor Kubernetes — the container orchestration system that schedules and manages containers across a cluster of machines — we use Helm charts:
# Simplified Helm values.yaml example
replicaCount:
agentCore: 2
groundingSkill: 3
translationSkill: 3
images:
repository:
agentCore: agent-core
groundingSkill: grounding-skill
translationSkill: translation-skill
tag: latest
pullPolicy: Always
service:
type: ClusterIP
port: 80Notice the replica counts: the grounding skill runs three copies because search is heavier than translation. You can scale each skill independently based on its actual load, which you can’t do with a monolith.
What changed
The translation and grounding teams can develop without stepping on each other. If grounding gets 10× the traffic, it can get 10× the containers while translation stays at one. Each skill can choose its own language-model client without sharing a dependency tree. A new skill can deploy without restarting the core, and a skill fix can ship without touching the rest.
Failure is isolated too: grounding can crash while translation and the core stay up. SSE sends status, thinking, and result updates as they happen. Each skill can be tested alone with mock SSE responses before integration.
The parts that stayed hard
This architecture makes some things better and some things worse. Here’s what got harder:
More Moving Parts
You now have N services instead of one. Each needs its own monitoring, logging, and deployment pipeline. I addressed this with automated deploys, centralized logging, and service discovery — but there’s no way around the fact that you’re now operating a distributed system, and distributed systems are harder to reason about than single processes.
Consistency Across Skills
When two different teams build two different skills, they drift. One uses status events; the other uses progress. One sends timestamps in UTC; the other in local time. The fix is boring but necessary: strict interface contracts, shared utility code, and regular cross-team code review.
Error Handling Gets Distributed
# Example of error handling in the agent core
async def execute_skill_with_fallback(self, primary_skill, fallback_skill, input_data):
"""Execute a skill with fallback if it fails."""
try:
async for update in self.execute_skill(primary_skill, input_data):
yield update
except SkillExecutionError as e:
logger.error(f"Primary skill {primary_skill} failed: {str(e)}")
logger.info(f"Falling back to {fallback_skill}")
# Notify the client of the fallback
yield {
"type": "status",
"data": {"status": f"Falling back to alternative approach"}
}
# Execute fallback skill
async for update in self.execute_skill(fallback_skill, input_data):
yield updateWhen a skill fails, the core needs a fallback plan. That fallback has to be designed and tested — it doesn’t fall out of the architecture for free.
State Across Service Boundaries
When the core calls Skill A and then Skill B, Skill B might need context from Skill A’s result. That shared state has to live somewhere outside either process. I used Redis — an in-memory data store — with automatic expiration so stale sessions clean themselves up:
class SharedStateManager:
"""Manages shared state across skills."""
def __init__(self, redis_url):
self.redis = aioredis.from_url(redis_url)
async def set_state(self, session_id, key, value, ttl=3600):
"""Set a state value with expiration."""
state_key = f"session:{session_id}:{key}"
await self.redis.set(state_key, json.dumps(value), ex=ttl)
async def get_state(self, session_id, key):
"""Get a state value."""
state_key = f"session:{session_id}:{key}"
value = await self.redis.get(state_key)
return json.loads(value) if value else NoneWatching it run
With services spread across processes, you need to see what’s happening. Here’s a monitoring middleware for FastAPI that tracks request counts and durations:
# Example monitoring middleware for FastAPI
@app.middleware("http")
async def add_performance_monitoring(request, call_next):
start_time = time.time()
# Track request by type
metrics.increment(f"requests.{request.url.path}")
# Execute the request
response = await call_next(request)
# Record duration
duration = time.time() - start_time
metrics.timing(f"request_duration.{request.url.path}", duration)
# Track response codes
metrics.increment(f"responses.{response.status_code}")
return responseStart modular with the third skill
Skill one and two feel fine in a monolith, so the point when modularity pays off is easy to miss. What I’d tell myself six months ago: start modular on skill three, before every new feature makes the monolith harder to change.
That means a thin orchestration core routing to independently deployable skills over SSE. It’s more pieces to manage, and the operational complexity is real. But the trade makes sense: you pay the setup cost once, then get it back every time you add a capability without touching existing skills or understanding the whole system. For any team building AI agents that will accumulate capabilities over time, that’s the test that matters.