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

python/src/uagents/asgi.py

@@ -7,12 +7,12 @@
 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
 from uagents.dispatch import dispatcher
-from uagents.envelope import Envelope
+from uagents.envelope import Envelope, EnvelopeHistory
 from uagents.models import ErrorMessage
 from uagents.utils import get_logger
 
@@ -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.
+        """
+        response = EnvelopeHistory(
+            envelopes=dispatcher.received_messages + self._dispenser.sent_messages
+        ).model_dump_json()
+        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

@@ -13,7 +13,7 @@
 from uagents.config import DEFAULT_ENVELOPE_TIMEOUT_SECONDS
 from uagents.crypto import Identity, is_user_address
 from uagents.dispatch import JsonStr, dispatcher
-from uagents.envelope import Envelope
+from uagents.envelope import Envelope, EnvelopeHistory, EnvelopeHistoryEntry
 from uagents.models import Model
 from uagents.resolver import GlobalResolver, Resolver
 from uagents.utils import get_logger
@@ -72,6 +72,7 @@ def __init__(self):
         self._envelopes: asyncio.Queue[
             Tuple[Envelope, List[str], asyncio.Future, bool]
         ] = asyncio.Queue()
+        self.sent_messages: EnvelopeHistory = []
 
     def add_envelope(
         self,
@@ -104,6 +105,11 @@ async def run(self):
                     sync=sync,
                 )
                 response_future.set_result(result)
+                env_dict = env.model_dump()
+                env_dict["payload"] = env.decode_payload()
+                env_dict["timestamp"] = int(time())
+
+                self.sent_messages.append(EnvelopeHistoryEntry(**env_dict))
             except Exception as err:
                 LOGGER.error(f"Failed to send envelope: {err}")
 

python/src/uagents/dispatch.py

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

python/src/uagents/envelope.py

@@ -3,11 +3,13 @@
 import base64
 import hashlib
 import struct
-from typing import Callable, Optional
+import time
+from typing import Callable, List, Optional
 
-from pydantic import UUID4, BaseModel, ConfigDict
+from pydantic import UUID4, BaseModel, ConfigDict, Field, field_serializer
 from uagents.crypto import Identity
-from uagents.dispatch import JsonStr
+
+JsonStr = str
 
 
 class Envelope(BaseModel):
@@ -109,3 +111,28 @@ def _digest(self) -> bytes:
         if self.nonce is not None:
             hasher.update(struct.pack(">Q", self.nonce))
         return hasher.digest()
+
+
+class EnvelopeHistoryEntry(BaseModel):
+    timestamp: int = Field(default_factory=lambda: int(time.time()))
+    version: int
+    sender: str
+    target: str
+    session: UUID4
+    schema_digest: str
+    protocol_digest: Optional[str] = None
+    payload: Optional[str] = None
+
+    @field_serializer("session")
+    def serialize_dt(self, session: UUID4, _info):
+        return str(session)
+
+
+class EnvelopeHistory(BaseModel):
+    envelopes: List[EnvelopeHistoryEntry]
+
+    @field_serializer("envelopes", when_used="json")
+    def serialize_envelopes_in_order(
+        self, envelopes: List[EnvelopeHistoryEntry], _info
+    ):
+        return sorted(envelopes, key=lambda e: e.timestamp)