feat(core): add message endpoint for inspector

python/src/uagents/agent.py

@@ -384,7 +384,11 @@ def __init__(
 
         if not self._use_mailbox:
             self._server = ASGIServer(
-                self._port, self._loop, self._queries, logger=self._logger
+                self._port,
+                self._loop,
+                self._queries,
+                self._dispenser,
+                logger=self._logger,
             )
 
         # define default error message handler
@@ -851,13 +855,13 @@ def include(self, protocol: Protocol, publish_manifest: Optional[bool] = False):
             if schema_digest in self._signed_message_handlers:
                 raise RuntimeError("Unable to register duplicate message handler")
             if schema_digest in protocol.signed_message_handlers:
-                self._signed_message_handlers[schema_digest] = (
-                    protocol.signed_message_handlers[schema_digest]
-                )
+                self._signed_message_handlers[
+                    schema_digest
+                ] = protocol.signed_message_handlers[schema_digest]
             elif schema_digest in protocol.unsigned_message_handlers:
-                self._unsigned_message_handlers[schema_digest] = (
-                    protocol.unsigned_message_handlers[schema_digest]
-                )
+                self._unsigned_message_handlers[
+                    schema_digest
+                ] = protocol.unsigned_message_handlers[schema_digest]
             else:
                 raise RuntimeError("Unable to lookup up message handler in protocol")
 
@@ -1147,7 +1151,10 @@ def __init__(
         self._port = port or 8000
         self._queries: Dict[str, asyncio.Future] = {}
         self._logger = get_logger("bureau", log_level)
-        self._server = ASGIServer(self._port, self._loop, self._queries, self._logger)
+        self._dispenser = Dispenser()
+        self._server = ASGIServer(
+            self._port, self._loop, self._queries, self._dispenser, self._logger
+        )
         self._use_mailbox = False
 
         if agents is not None:

python/src/uagents/asgi.py

@@ -7,7 +7,7 @@
 import pydantic
 import uvicorn
 from requests.structures import CaseInsensitiveDict
-from uagents.communication import enclose_response_raw
+from uagents.communication import Dispenser, enclose_response_raw
 from uagents.config import RESPONSE_TIME_HINT_SECONDS
 from uagents.context import ERROR_MESSAGE_DIGEST
 from uagents.crypto import is_user_address
@@ -44,6 +44,7 @@ def __init__(
         port: int,
         loop: asyncio.AbstractEventLoop,
         queries: Dict[str, asyncio.Future],
+        dispenser: Dispenser,
         logger: Optional[Logger] = None,
     ):
         """
@@ -58,6 +59,7 @@ def __init__(
         self._port = int(port)
         self._loop = loop
         self._queries = queries
+        self._dispenser = dispenser
         self._logger = logger or get_logger("server")
         self._server = None
 
@@ -149,6 +151,26 @@ async def handle_missing_content_type(self, headers: CaseInsensitiveDict, send):
                 }
             )
 
+    async def handle_get_messages(self, send):
+        """
+        Handle retrieval of stored messages.
+        """
+        messages = dispatcher.received_messages + self._dispenser.sent_messages
+        messages.sort(key=lambda x: x["timestamp"])
+        response = {"messages": [msg for msg in messages]}
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [
+                    [b"content-type", b"application/json"],
+                ],
+            }
+        )
+        await send(
+            {"type": "http.response.body", "body": json.dumps(response).encode()}
+        )
+
     async def serve(self):
         """
         Start the server.
@@ -173,6 +195,13 @@ async def __call__(self, scope, receive, send):  #  pylint: disable=too-many-bra
 
         assert scope["type"] == "http"
 
+        request_method = scope["method"]
+        path = scope["path"]
+
+        if request_method == "GET" and path == "/messages":
+            await self.handle_get_messages(send)
+            return
+
         if scope["path"] != "/submit":
             await send(
                 {
@@ -190,7 +219,6 @@ async def __call__(self, scope, receive, send):  #  pylint: disable=too-many-bra
 
         headers = CaseInsensitiveDict(scope.get("headers", {}))
 
-        request_method = scope["method"]
         if request_method == "HEAD":
             await self.handle_readiness_probe(headers, send)
             return

python/src/uagents/communication.py

@@ -72,6 +72,7 @@ def __init__(self):
         self._envelopes: asyncio.Queue[
             Tuple[Envelope, List[str], asyncio.Future, bool]
         ] = asyncio.Queue()
+        self.sent_messages = []
 
     def add_envelope(
         self,
@@ -104,6 +105,17 @@ async def run(self):
                     sync=sync,
                 )
                 response_future.set_result(result)
+
+                self.sent_messages.append({
+                    "timestamp": time(),
+                    "envelope": {
+                        "sender": env.sender,
+                        "target": env.target,
+                        "schema_digest": env.schema_digest,
+                        "payload": env.decode_payload(),
+                        "session": str(env.session),
+                    }
+                })
             except Exception as err:
                 LOGGER.error(f"Failed to send envelope: {err}")
 

python/src/uagents/dispatch.py

@@ -1,5 +1,6 @@
 import uuid
 from abc import ABC, abstractmethod
+from time import time
 from typing import Dict, Set
 
 JsonStr = str
@@ -24,6 +25,7 @@ class Dispatcher:
 
     def __init__(self):
         self._sinks: Dict[str, Set[Sink]] = {}
+        self.received_messages = []
 
     @property
     def sinks(self) -> Dict[str, Set[Sink]]:
@@ -56,5 +58,15 @@ async def dispatch(
         for handler in self._sinks.get(destination, set()):
             await handler.handle_message(sender, schema_digest, message, session)
 
+        self.received_messages.append({
+            "timestamp": time(),
+            "envelope": {
+                "sender": sender,
+                "target": destination,
+                "schema_digest": schema_digest,
+                "payload": message,
+                "session": str(session),
+            }
+        })
 
 dispatcher = Dispatcher()