Agent Discovery

This guide covers how to register your agent’s capabilities and discover other agents.

Registering Your Agent

Basic Registration

from acenta import AcentaClient

client = AcentaClient(api_key="...")

# Register a capability
await client.discovery.register_capability(
    agent_id="my-document-agent",
    capability="document.ocr",
    metadata={
        "supported_formats": ["pdf", "png", "jpg"],
        "max_file_size_mb": 50,
        "version": "1.0.0"
    }
)

Multiple Capabilities

An agent can have multiple capabilities:

capabilities = [
    ("document.ocr", {"formats": ["pdf", "png"]}),
    ("document.translate", {"languages": ["en", "es", "fr", "de"]}),
    ("document.summarize", {"max_length": 10000})
]

for capability, metadata in capabilities:
    await client.discovery.register_capability(
        agent_id="my-document-agent",
        capability=capability,
        metadata=metadata
    )

With Groups

Register in groups for logical organization:

await client.discovery.register_capability(
    agent_id="ocr-agent-us-east-1",
    capability="document.ocr",
    groups=["production", "us-east", "high-capacity"]
)

Capability Naming

Use a hierarchical naming convention:

Format	Example
domain.action	document.ocr
domain.subdomain.action	image.face.detect

Good capability names:

• document.ocr
• document.translate
• image.resize
• audio.transcribe
• payment.process

Health Management

Send Heartbeats

Keep your agent marked as healthy:

import asyncio

async def heartbeat_loop(client, agent_id):
    while True:
        await client.discovery.heartbeat(
            agent_id=agent_id,
            load=get_current_load(),  # 0.0 to 1.0
            metadata={
                "active_tasks": count_active_tasks(),
                "memory_mb": get_memory_usage()
            }
        )
        await asyncio.sleep(30)  # Every 30 seconds

Health Status

Agents are marked as:

Status	Description
healthy	Heartbeat received recently
degraded	Heartbeat late or high load
unhealthy	No heartbeat for 3+ intervals

Finding Agents

By Capability

# Find all agents with a capability
agents = await client.discovery.find_agents(
    capability="document.ocr"
)

for agent in agents:
    print(f"{agent.agent_id}: load={agent.load:.2f}")

With Routing Strategy

# Get the best agent based on strategy
agents = await client.discovery.find_agents(
    capability="document.ocr",
    routing_strategy="least_loaded",
    limit=1
)

target = agents[0] if agents else None

By Group

# Find agents in a specific group
agents = await client.discovery.find_agents(
    capability="document.ocr",
    group="us-east"
)

With Metadata Filter

# Find agents matching specific metadata
agents = await client.discovery.find_agents(
    capability="document.ocr",
    metadata_filter={
        "supported_formats": {"$contains": "pdf"},
        "max_file_size_mb": {"$gte": 100}
    }
)

Semantic Search

Use natural language to find agents:

results = await client.discovery.search(
    query="I need to convert audio to text in Spanish",
    top_k=5,
    min_score=0.6
)

for result in results:
    print(f"{result.capability.name}: {result.similarity_score:.2f}")
    print(f"  Agent: {result.agent_id}")
    print(f"  Metadata: {result.capability.metadata}")

Routing Strategies

Strategy	Best For
least_loaded	Load balancing
round_robin	Even distribution
random	Simple random selection
latency	Latency-sensitive workloads
trust	Security-sensitive operations
weighted	Custom weighting

Deregistration

Remove Specific Capability

await client.discovery.deregister_capability(
    agent_id="my-agent",
    capability="document.ocr"
)

Remove All Capabilities

await client.discovery.deregister(agent_id="my-agent")

Agent Lifecycle Pattern

class MyAgent:
    def __init__(self, client, agent_id):
        self.client = client
        self.agent_id = agent_id
        self.running = False

    async def start(self):
        self.running = True

        # Register capabilities
        await self.register()

        # Start heartbeat
        asyncio.create_task(self.heartbeat_loop())

        # Process messages
        await self.process_messages()

    async def register(self):
        await self.client.discovery.register_capability(
            agent_id=self.agent_id,
            capability="my.capability"
        )

    async def heartbeat_loop(self):
        while self.running:
            await self.client.discovery.heartbeat(
                agent_id=self.agent_id,
                load=self.get_load()
            )
            await asyncio.sleep(30)

    async def stop(self):
        self.running = False
        await self.client.discovery.deregister(
            agent_id=self.agent_id
        )

Next Steps

• Messaging Patterns - Communicate with discovered agents
• Building Workflows - Use agents in workflows
• API Reference - Complete API documentation