Architecting Extensible AI Agents - A Modular Core with Pluggable Skills and SSE Communication

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.

Here’s the architecture, the wrong turns, and why SSE beat the alternatives for this specific job.

Why One Big Agent Stopped Working

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.

The specific problems:

1. Adding a translation feature broke the search feature. The translation code imported a newer version of the language-model library, which shifted a dependency the search feature relied on. Took me an afternoon to untangle. That’s tight coupling — when two pieces of code can’t change independently without surprising each other.

2. Deploying a small fix meant restarting everything. Even a one-line change to the grounding logic required the full agent to go down and come back up. If a user was mid-request, too bad. That’s a deployment problem you earn by putting everything in one box.

3. Different skills wanted different technology. The translation skill really wanted the latest language model client library. The grounding skill needed a search API client. The monolith forced them to share the same environment, same versions, same everything. That’s technology lock-in — the format of your system dictates what tools you can use 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.

The Modular Bet: Core, Skills, and a Communication Layer

The alternative splits the system into three pieces:

1. An agent core — handles routing (which skill should handle this request?), session state (what’s the user’s context?), and the public API. It doesn’t do any of the actual work.
2. Skills — standalone services, each doing one thing (translation, grounding, search). Each owns its own dependencies, can be deployed independently, and exposes a single endpoint the core calls.
3. A communication layer — how the core talks to skills and how results stream back to the client. This is where SSE lives.

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: the skill fires status → thinking → result events as it works, and the core streams each one to the client immediately. The user sees progress, not a spinner.

I initially tried having skills return a single JSON blob after finishing. That’s what caused the deadlock in the opening story — the core couldn’t show progress because it had to wait for the full response. SSE fixed that by letting skills emit events as they happen rather than bundling everything at the end.

SSE for Real-Time Communication

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 went with SSE. Here’s why:

1. It’s simpler. SSE is unidirectional — server to client only. That’s actually what you want here: the client sends one request (a POST), then listens. You don’t need a two-way socket. Less code, fewer edge cases.
2. It rides plain HTTP. No upgrade negotiation, no WebSocket-aware proxy needed. It works through every reverse proxy, load balancer, and CDN without special configuration.
3. Browsers reconnect automatically. If the connection drops, the browser re-establishes it. You don’t write reconnection logic.
4. Named events. SSE supports typed events (event: status, event: result), which maps cleanly to the different update types a skill produces.

The implementation includes custom SSE handling on both client and server sides:

curl -N -X POST http://localhost:8081/execute \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -d '{"statement": "The sky is blue"}'

The Translation Skill: A Complete Example

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_lang

The 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.

Integration with MCP Protocol

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.

Client-Side: Listening to the Stream

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.

Deploying It

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: bridge

For 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: 80

Notice 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 This Architecture Gets You

• Independent development. The translation team and the grounding team don’t step on each other.
• Independent scaling. If grounding gets 10× the traffic, give it 10× the containers. Translation stays at one.
• Technology choice per skill. Grounding can use one language model client; translation can use another. They don’t share a dependency tree.
• Incremental deployment. Add a new skill without restarting the core. Deploy a skill fix without touching anything else.
• Failure isolation. The grounding skill crashing doesn’t take down translation. The core stays up.
• Streaming updates. SSE pushes progress to the client as it happens — status, thinking, result — instead of making the user stare at a loading bar.
• Simplified testing. Test each skill in isolation with mock SSE responses before integrating.

The Hard Parts

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 update

When 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 None

Watching 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 response

The Takeaway

The modular architecture — a thin core routing to independent skills over SSE — replaced a monolith that got harder to change with every new feature. It’s more pieces to manage, and the operational complexity is real. But the trade makes sense: you pay the complexity cost once when you set up the infrastructure, and you get it back every time you add a new capability without touching the existing ones.

What I’d tell myself six months ago: start modular on skill three. Skill one and two feel fine in a monolith, and by the time you feel the pain, you’re already in it.

The combination of a strong orchestration core, independently deployable skills, and SSE for streaming communication gives you a foundation that can grow without requiring you to understand the whole system every time you make a change. For any team building AI agents that will accumulate capabilities over time, that’s the test that matters.