feat: add broadcast to agents via protocol

src/uagents/config.py

@@ -27,6 +27,7 @@ class AgentNetwork(Enum):
 AGENT_NETWORK = AgentNetwork.FETCHAI_TESTNET
 
 AGENTVERSE_URL = "https://agentverse.ai"
+ALMANAC_API_URL = "https://staging.agentverse.ai/v1/almanac/"
 MAILBOX_POLL_INTERVAL_SECONDS = 1.0
 
 DEFAULT_ENVELOPE_TIMEOUT_SECONDS = 30

src/uagents/context.py

@@ -1,4 +1,5 @@
 from __future__ import annotations
+
 import asyncio
 import logging
 import uuid
@@ -7,14 +8,15 @@
 from typing import Dict, Set, Optional, Callable, Any, Awaitable, Type, TYPE_CHECKING
 
 import aiohttp
+import requests
 from cosmpy.aerial.client import LedgerClient
 from cosmpy.aerial.wallet import LocalWallet
 
-from uagents.config import DEFAULT_ENVELOPE_TIMEOUT_SECONDS
+from uagents.config import ALMANAC_API_URL, DEFAULT_ENVELOPE_TIMEOUT_SECONDS
 from uagents.crypto import Identity
 from uagents.dispatch import JsonStr, dispatcher
 from uagents.envelope import Envelope
-from uagents.models import Model, ErrorMessage
+from uagents.models import ErrorMessage, Model
 from uagents.resolver import Resolver
 from uagents.storage import KeyValueStore
 
@@ -97,22 +99,61 @@ def get_message_protocol(self, message_schema_digest) -> Optional[str]:
                     return protocol_digest
         return None
 
+    def get_agents_by_protocol(self, protocol_digest: str) -> list:
+        if not isinstance(protocol_digest, str) or not protocol_digest.startswith(
+            "proto:"
+        ):
+            self.logger.error(f"Invalid protocol digest: {protocol_digest}")
+            return []
+        response = requests.post(
+            url=ALMANAC_API_URL + "search",
+            json={"text": protocol_digest[6:]},
+            timeout=DEFAULT_ENVELOPE_TIMEOUT_SECONDS,
+        )
+        if response.status_code == 200:
+            data = response.json()
+            return [agent["address"] for agent in data if agent["status"] == "local"]
+        return []
+
     async def send(
         self,
         destination: str,
         message: Model,
         timeout: Optional[int] = DEFAULT_ENVELOPE_TIMEOUT_SECONDS,
     ):
         schema_digest = Model.build_schema_digest(message)
-        await self.send_raw(
+        await self._send_raw(
             destination,
             message.json(),
             schema_digest,
             message_type=type(message),
             timeout=timeout,
         )
 
-    async def send_raw(
+    async def broadcast(
+        self,
+        destination: str,
+        message: Model,
+        timeout: Optional[int] = DEFAULT_ENVELOPE_TIMEOUT_SECONDS,
+    ):
+        agents = self.get_agents_by_protocol(destination)
+        if not agents:
+            self.logger.error(f"No active agents found for: {destination}")
+            return
+        schema_digest = Model.build_schema_digest(message)
+        for address in agents:
+            try:
+                await self._send_raw(
+                    address,
+                    message.json(),
+                    schema_digest,
+                    message_type=type(message),
+                    timeout=timeout,
+                )
+            except Exception as e:
+                self.logger.error(f"Error sending message to {address}: {e}")
+
+    async def _send_raw(
         self,
         destination: str,
         json_message: JsonStr,