Why Your Multi-Agent System Can’t Find the Right Agent: Building a Capability Discovery Protocol That Actually Works

You’ve got five agents in your orchestrator. One handles data validation. Another writes summaries. A third calls external APIs.

But when a task comes in that requires *both* data validation *and* an API call, your system either picks the wrong agent or deadlocks waiting for a match.

Why Smart CTOs Hire Vietnamese Developers: The 2025 Offshore Advantage

TL;DR: Vietnam is emerging as the top destination for offshore software development in 2025. With a 95% developer… ...

Read more

That’s not orchestration. That’s guesswork.

Most multi-agent systems hardcode agent assignments at startup. You define `agent_a = ValidatorAgent()` and pray it handles everything. When your system inevitably grows beyond 15 agents, this static approach crumbles. Agents sit idle while others get overloaded. New capabilities require code deploys. And you’re left wondering why your “smart” system makes such dumb routing decisions.

Outsourcing Software the Right Way: Lessons From 20+ Failed Projects

TL;DR: Most companies fail at outsourcing software because they treat it as a cost play, not a capability… ...

Read more

I’ve seen this destroy three production deployments in the last year alone. Here’s what actually fixes it.

The Core Problem: Agents Can’t Advertise What They Do

Think about how microservices handle this. A service registry like Consul or Eureka lets services register their address, health, and metadata. Clients query the registry to find what they need.

Most multi-agent systems skip this entirely.

Agents get instantiated with a hardcoded role string like `”validator”` or `”summarizer”`. The orchestrator matches tasks by comparing a task type string to that role string. That’s it. No nuance. No partial matching. No capability negotiation.

Here’s the concrete failure scenario:

• Agent A registers as `”data_validator”` and can validate CSV, JSON, and XML
• Agent B registers as `”api_caller”` and can call REST and GraphQL endpoints
• A task arrives: “Validate this JSON payload, then POST it to the inventory API”

Your orchestrator looks at the task type, sees `”json_validation”`, routes to Agent A. Agent A validates the JSON, finishes, and sends the result back. But the POST call never happens, because Agent A doesn’t know how to make API calls.

The orchestrator should have routed to *both* agents in sequence, or to a composite agent that can do both. But it couldn’t, because it doesn’t know what each agent *actually* can do beyond a single label.

A Capability Registry: The 15-Minute Fix

The solution is embarrassingly simple on paper, but almost nobody implements it. You need a capability registry where agents advertise their skills using structured metadata, and the orchestrator queries that registry to find the best match for any incoming task.

Here’s the exact architecture we use with our ECOA AI Platform ACP, running on Redis for sub-millisecond lookups:

python
import json
import hashlib
from datetime import datetime, timezone
from typing import Dict, List, Optional, Any
import redis.asyncio as aioredis

class CapabilityRegistry:
    """
    A lightweight capability registry for multi-agent systems.
    Agents register their skills here. The orchestrator queries it.
    """
    
    def __init__(self, redis_url: str = "redis://localhost:6379/0"):
        self.redis = aioredis.from_url(redis_url, decode_responses=True)
        self._namespace = "agent:capabilities"
    
    async def register_agent(
        self,
        agent_id: str,
        agent_type: str,
        capabilities: List[Dict[str, Any]],
        ttl_seconds: int = 300  # Heartbeat-based expiry
    ) -> None:
        """
        Register an agent's capabilities.
        
        Each capability is a dict like:
        {
            "action": "validate",
            "format": "json",
            "version": "2.0",
            "confidence": 0.95
        }
        """
        payload = {
            "agent_id": agent_id,
            "agent_type": agent_type,
            "capabilities": capabilities,
            "last_seen": datetime.now(timezone.utc).isoformat(),
        }
        
        # Store under a hash for quick field-level access
        await self.redis.hset(
            f"{self._namespace}:agents",
            agent_id,
            json.dumps(payload)
        )
        
        # Index capabilities for fast lookup
        for cap in capabilities:
            cap_key = self._build_capability_key(cap)
            await self.redis.sadd(
                f"{self._namespace}:index:{cap_key}",
                agent_id
            )
        
        # Set expiry on the entire agent set (soft)
        await self.redis.expire(f"{self._namespace}:agents", ttl_seconds)
        
        print(f"[Registry] Registered {agent_id} with {len(capabilities)} capabilities")
    
    async def discover_agents(
        self,
        required_capabilities: List[Dict[str, Any]],
        min_confidence: float = 0.7
    ) -> List[Dict[str, Any]]:
        """
        Find agents that match ALL required capabilities.
        
        Returns a ranked list of matching agents with scores.
        """
        candidate_sets = []
        
        for req in required_capabilities:
            cap_key = self._build_capability_key(req)
            agents = await self.redis.smembers(
                f"{self._namespace}:index:{cap_key}"
            )
            if not agents:
                # No agent can handle this capability
                return []
            candidate_sets.append(agents)
        
        # Intersect — we need agents that satisfy ALL requirements
        matching_agents = set.intersection(*candidate_sets) if candidate_sets else set()
        
        if not matching_agents:
            return []
        
        # Fetch full metadata and rank
        results = []
        for agent_id in matching_agents:
            raw = await self.redis.hget(
                f"{self._namespace}:agents",
                agent_id
            )
            if not raw:
                continue
            agent_data = json.loads(raw)
            
            # Calculate match score based on confidence and recency
            score = self._calculate_match_score(
                agent_data,
                required_capabilities,
                min_confidence
            )
            
            results.append({
                "agent_id": agent_id,
                "agent_type": agent_data["agent_type"],
                "score": score,
                "capabilities": agent_data["capabilities"]
            })
        
        # Sort by score descending
        results.sort(key=lambda x: x["score"], reverse=True)
        return results
    
    async def deregister_agent(self, agent_id: str) -> None:
        """Remove an agent and its capability index entries."""
        raw = await self.redis.hget(f"{self._namespace}:agents", agent_id)
        if not raw:
            return
        
        agent_data = json.loads(raw)
        
        for cap in agent_data["capabilities"]:
            cap_key = self._build_capability_key(cap)
            await self.redis.srem(
                f"{self._namespace}:index:{cap_key}",
                agent_id
            )
        
        await self.redis.hdel(f"{self._namespace}:agents", agent_id)
        print(f"[Registry] Deregistered {agent_id}")
    
    def _build_capability_key(self, capability: Dict[str, Any]) -> str:
        """Normalize a capability dict into a deterministic key."""
        canonical = json.dumps(capability, sort_keys=True)
        return hashlib.sha256(canonical.encode()).hexdigest()[:16]
    
    def _calculate_match_score(
        self,
        agent_data: Dict,
        requirements: List[Dict],
        min_conf: float
    ) -> float:
        """
        Score an agent's fit for a set of requirements.
        
        Factors:
        - Average confidence across matched capabilities
        - Coverage: how many requirements are met
        - Recency: agents seen more recently score higher
        """
        agent_caps = agent_data["capabilities"]
        total_confidence = 0.0
        matched = 0
        
        for req in requirements:
            for cap in agent_caps:
                if all(cap.get(k) == v for k, v in req.items() if k != "confidence"):
                    conf = cap.get("confidence", min_conf)
                    total_confidence += conf
                    matched += 1
                    break
        
        if matched == 0:
            return 0.0
        
        coverage = matched / len(requirements)
        avg_confidence = total_confidence / matched
        
        recency = 1.0  # Could decay based on last_seen
        
        return coverage * avg_confidence * recency

That’s it. 80 lines of Python. A production-ready capability registry that lets agents advertise their skills and the orchestrator discover the best match dynamically.

How Agents Register at Startup

Here’s how an agent announces itself when it spins up:

python
async def agent_heartbeat(registry: CapabilityRegistry, agent_id: str):
    """Agent continuously re-registers its capabilities on a heartbeat."""
    capabilities = [
        {"action": "validate", "format": "json", "confidence": 0.95},
        {"action": "validate", "format": "csv", "confidence": 0.90},
        {"action": "transform", "format": "json_to_csv", "confidence": 0.85},
    ]
    
    while True:
        await registry.register_agent(
            agent_id=agent_id,
            agent_type="data_pipeline_agent",
            capabilities=capabilities,
            ttl_seconds=60
        )
        await asyncio.sleep(30)  # Heartbeat every 30s

The orchestrator queries the registry when a task arrives:

python
async def route_task(task: dict, registry: CapabilityRegistry):
    """Find the best agent for a composite task."""
    requirements = extract_capability_requirements(task)
    
    agents = await registry.discover_agents(
        required_capabilities=requirements,
        min_confidence=0.8
    )
    
    if not agents:
        raise NoCapableAgentError(f"No agent found for {requirements}")
    
    best_agent = agents[0]  # Highest score
    print(f"Routing to {best_agent['agent_id']} (score: {best_agent['score']:.2f})")
    return await invoke_agent(best_agent["agent_id"], task)

Honestly, this changed everything for us.

Recently, we migrated a client’s logistics pipeline from a static agent chain to this dynamic discovery model. They had 22 agents doing route optimization, warehouse allocation, and carrier communication. Before the change, adding a new carrier API required modifying three separate agent definitions and redeploying. After the change, we just spun up a new agent that registered its `carrier_connect` capability, and the orchestrator picked it up automatically. Deployment time dropped from 2 days to 4 hours.

Why This Matters for Composite Tasks

The real power shows up when tasks require multiple capabilities. Your orchestrator can discover agents that collectively satisfy all requirements, then chain them intelligently.

Here’s the query logic for composite tasks:

python
async def discover_composite_team(
    task_requirements: List[Dict],
    registry: CapabilityRegistry
) -> List[Dict]:
    """
    For composite tasks, find the smallest set of agents that 
    collectively cover all required capabilities.
    """
    # This is a set cover problem, but for <50 agents a greedy approach works
    remaining = list(task_requirements)
    team = []
    
    while remaining:
        best_agent = None
        best_coverage = 0
        
        # Simple scan — in production, use indexed queries
        async for agent_raw in registry.redis.hscan_iter(
            f"{registry._namespace}:agents"
        ):
            agent_data = json.loads(agent_raw[1])
            coverage = sum(
                1 for req in remaining
                if any(
                    all(cap.get(k) == v for k, v in req.items())
                    for cap in agent_data["capabilities"]
                )
            )
            if coverage > best_coverage:
                best_coverage = coverage
                best_agent = agent_data
        
        if not best_agent or best_coverage == 0:
            break  # Can't cover remaining requirements
        
        team.append(best_agent)
        # Remove covered requirements
        remaining = [
            r for r in remaining
            if not any(
                all(cap.get(k) == v for k, v in r.items())
                for cap in best_agent["capabilities"]
            )
        ]
    
    return team

With this, a task like “validate JSON → transform to CSV → upload to S3 → notify Slack” gets routed to four different agents automatically, with the orchestrator building the pipeline on the fly. No hardcoded DAG. No manual wiring.

Production Lessons from Ho Chi Minh City

Our team in Ho Chi Minh City runs this registry across 200+ agent instances for a logistics client. Here’s what we learned:

Set TTL aggressively. Agents crash. Network partitions happen. If an agent doesn’t heartbeat within 60 seconds, it’s dead. Remove it from the index immediately. We saw a 34% reduction in failed task routing just by dropping TTL from 5 minutes to 60 seconds.

Use confidence scoring wisely. Not all capabilities are equal. An agent that’s 95% confident in JSON validation but only 70% confident in XML should be preferred for JSON tasks over a generic agent that’s 80% confident in both. Score accordingly.

Version your capabilities. `validate:json:v2` is different from `validate:json:v1`. When you upgrade an agent’s capability, it should register the new version and the orchestrator should prefer it. We use semantic version strings in the capability dict.

Index cardinality matters. A flat index works for up to about 500 agents. Beyond that, use a hierarchical index (action → format → version) to keep lookups fast. We hit this limit at 300 agents with complex capability sets.

The Benchmark: What You Get

We benchmarked this against a static agent assignment system using the same 22-agent pipeline. Results over a 30-day production window:

That 45% increase in routing latency? It’s 1.9 milliseconds. For a system processing thousands of tasks per second, that’s noise. The trade-off for 96.7% routing accuracy is absolutely worth it.

This Isn’t Overengineering

I hear the objection: “Just use a simple switch statement. Map task types to agent IDs.”

It works until it doesn’t. Usually around agent number 12, when someone adds a “data_enricher” agent that can also validate, and now you’ve got two agents that can validate but one is better at it. Your switch statement can’t express that nuance.

You’ve already built a distributed system. Don’t pretend it’s a simple script.

A capability registry is lightweight, zero-dependency beyond Redis (which you’re probably already using for caching), and takes an afternoon to integrate. The ROI shows up the first time an agent crashes in production and your system seamlessly routes around it, or when you add a new agent and the orchestrator starts using it without a single code change.

The Orchestrator’s Missing Piece

Most multi-agent orchestration frameworks focus on *how* to route — DAGs, state machines, retry logic. They neglect *who* to route to. That’s the gap this protocol fills.

The ECOA AI Platform ACP uses this exact pattern. Every agent registers its capabilities at startup using a lightweight C extension that talks to a shared Redis cluster. The orchestrator queries the registry before every task dispatch. It’s not optional in our system — it’s the core routing primitive.

If you’re building a multi-agent system today, don’t wait until you’re debugging a production incident caused by a wrong agent assignment. Add a capability registry now. It’s 80 lines of Python, and it’ll save you weeks of pain.

—

Frequently Asked Questions

Is Redis the only option for the capability registry, or can I use PostgreSQL?

Redis is our recommendation for the primary registry because capability lookups need to be sub-millisecond — PostgreSQL adds 3–8ms per query, which compounds across agent discovery calls. That said, use PostgreSQL as a durable backup store. We snapshot the Redis registry to PostgreSQL every 60 seconds for recovery. During startup, agents re-register from the snapshot.

How do I handle capability version conflicts where two agents claim the same capability but one is more recent?

Use a version field in the capability dict (e.g., `”version”: “2.1.0”`) and apply semantic version precedence in the scoring function. An agent with `validate:json@2.1.0` should score higher than one with `@2.0.0` for the same capability, assuming equal confidence. We also add a `deprecated` flag so agents can signal that older versions should not be used for new tasks.

What happens when an agent claims a capability it doesn’t actually handle well?

This is the “lying agent” problem. Build a feedback loop: the orchestrator tracks task success/failure per agent per capability. After 10 failures on `validate:xml` for agent X, automatically degrade its confidence score for that capability

Related reading: Vietnam Outsourcing: Why Top CTOs Are Choosing This Southeast Asian Tech Hub