refactor(core): context stage 2

python/examples/08-local-network-interaction/sync_sender.py

@@ -1,7 +1,7 @@
 import asyncio
 
 from uagents import Model
-from uagents.context import send_sync_message
+from uagents.communication import send_sync_message
 
 RECIPIENT_ADDRESS = (
     "test-agent://agent1q2kxet3vh0scsf0sm7y2erzz33cve6tv5uk63x64upw5g68kr0chkv7hw50"

python/examples/17-stateful-communication/agent3.py

@@ -7,8 +7,7 @@
     ChitChatDialogueMessage,
     ConcludeChitChatDialogue,
 )
-from uagents import Agent
-from uagents.context import Context
+from uagents import Agent, Context
 
 CHAT_AGENT_ADDRESS = "agent1qwvecfwc255pfqqwtjznh9qqk6skl77xc6fzw8mr3ppfex32sr0kcad62n4"
 

python/src/uagents/agent.py

@@ -4,14 +4,26 @@
 import functools
 import logging
 import uuid
-from typing import Any, Coroutine, Dict, List, Optional, Set, Tuple, Type, Union
+from typing import (
+    Any,
+    Callable,
+    Coroutine,
+    Dict,
+    List,
+    Optional,
+    Set,
+    Tuple,
+    Type,
+    Union,
+)
 
 import requests
 from cosmpy.aerial.client import LedgerClient
 from cosmpy.aerial.wallet import LocalWallet, PrivateKey
 from cosmpy.crypto.address import Address
 from pydantic import ValidationError
 from uagents.asgi import ASGIServer
+from uagents.communication import Dispenser, MsgDigest
 from uagents.config import (
     AVERAGE_BLOCK_INTERVAL,
     LEDGER_PREFIX,
@@ -20,16 +32,16 @@
     REGISTRATION_RETRY_INTERVAL_SECONDS,
     REGISTRATION_UPDATE_INTERVAL_SECONDS,
     TESTNET_PREFIX,
-    get_logger,
     parse_agentverse_config,
     parse_endpoint_config,
 )
 from uagents.context import (
     Context,
     EventCallback,
+    ExternalContext,
+    InternalContext,
     IntervalCallback,
     MessageCallback,
-    MsgDigest,
 )
 from uagents.crypto import Identity, derive_key_from_seed, is_user_address
 from uagents.dispatch import JsonStr, Sink, dispatcher
@@ -44,6 +56,7 @@
 from uagents.protocol import Protocol
 from uagents.resolver import GlobalResolver, Resolver
 from uagents.storage import KeyValueStore, get_or_create_private_keys
+from uagents.utils import get_logger
 
 
 async def _run_interval(func: IntervalCallback, ctx: Context, period: float):
@@ -92,6 +105,87 @@ async def _send_error_message(ctx: Context, destination: str, msg: ErrorMessage)
     await ctx.send(destination, msg)
 
 
+class AgentRepresentation:
+    """
+    Represents an agent in the context of a message.
+
+    Attributes:
+        _address (str): The address of the agent.
+        _name (Optional[str]): The name of the agent.
+        _signing_callback (Callable): The callback for signing messages.
+
+    Properties:
+        name (str): The name of the agent.
+        address (str): The address of the agent.
+        identifier (str): The agent's address and network prefix.
+
+    Methods:
+        sign_digest(data: bytes) -> str: Sign the provided data with the agent's identity.
+    """
+
+    def __init__(
+        self,
+        address: str,
+        name: Optional[str],
+        signing_callback: Callable,
+    ):
+        """
+        Initialize the AgentRepresentation instance.
+
+        Args:
+            address (str): The address of the context.
+            name (Optional[str]): The optional name associated with the context.
+            signing_callback (Callable): The callback for signing messages.
+        """
+        self._address = address
+        self._name = name
+        self._signing_callback = signing_callback
+
+    @property
+    def name(self) -> str:
+        """
+        Get the name associated with the context or a truncated address if name is None.
+
+        Returns:
+            str: The name or truncated address.
+        """
+        if self._name is not None:
+            return self._name
+        return self._address[:10]
+
+    @property
+    def address(self) -> str:
+        """
+        Get the address of the context.
+
+        Returns:
+            str: The address of the context.
+        """
+        return self._address
+
+    @property
+    def identifier(self) -> str:
+        """
+        Get the address of the agent used for communication including the network prefix.
+
+        Returns:
+            str: The agent's address and network prefix.
+        """
+        return TESTNET_PREFIX + "://" + self._address
+
+    def sign_digest(self, data: bytes) -> str:
+        """
+        Sign the provided data with the callback of the agent's identity.
+
+        Args:
+            data (bytes): The data to sign.
+
+        Returns:
+            str: The signature of the data.
+        """
+        return self._signing_callback(data)
+
+
 class Agent(Sink):
     """
     An agent that interacts within a communication environment.
@@ -121,7 +215,8 @@ class Agent(Sink):
         of incoming message.
         _queries (Dict[str, asyncio.Future]): Dictionary mapping query senders to their response
         Futures.
-        _dispatcher: The dispatcher for message handling.
+        _dispatcher: The dispatcher for internal handling/sorting of messages.
+        _dispenser: The dispatcher for external message handling.
         _message_queue: Asynchronous queue for incoming messages.
         _on_startup (List[Callable]): List of functions to run on agent startup.
         _on_shutdown (List[Callable]): List of functions to run on agent shutdown.
@@ -187,7 +282,6 @@ def __init__(
         """
         self._name = name
         self._port = port if port is not None else 8000
-        self._background_tasks: Set[asyncio.Task] = set()
         self._resolver = (
             resolve
             if resolve is not None
@@ -241,6 +335,7 @@ def __init__(
         self._replies: Dict[str, Dict[str, Type[Model]]] = {}
         self._queries: Dict[str, asyncio.Future] = {}
         self._dispatcher = dispatcher
+        self._dispenser = Dispenser()
         self._message_queue = asyncio.Queue()
         self._on_startup = []
         self._on_shutdown = []
@@ -255,20 +350,18 @@ def __init__(
         # keep track of supported protocols
         self.protocols: Dict[str, Protocol] = {}
 
-        self._ctx = Context(
-            self._identity.address,
-            self.identifier,
-            self._name,
-            self._storage,
-            self._resolver,
-            self._identity,
-            self._wallet,
-            self._ledger,
-            self._queries,
-            replies=self._replies,
+        self._ctx = InternalContext(
+            agent=AgentRepresentation(
+                address=self.address,
+                name=self._name,
+                signing_callback=self._identity.sign_digest,
+            ),
+            storage=self._storage,
+            ledger=self._ledger,
+            resolver=self._resolver,
+            dispenser=self._dispenser,
             interval_messages=self._interval_messages,
             wallet_messaging_client=self._wallet_messaging_client,
-            protocols=self.protocols,
             logger=self._logger,
         )
 
@@ -783,8 +876,6 @@ def include(self, protocol: Protocol, publish_manifest: Optional[bool] = False):
 
         if protocol.digest is not None:
             self.protocols[protocol.digest] = protocol
-            if self._ctx is not None:
-                self._ctx.update_protocols(protocol)
 
         if publish_manifest:
             self.publish_manifest(protocol.manifest())
@@ -866,34 +957,41 @@ def setup(self):
         """
         # register the internal agent protocol
         self.include(self._protocol)
+        self.start_message_dispenser()
         self._loop.run_until_complete(self._startup())
-        self.start_background_tasks()
+        self.start_message_receivers()
+        self.start_interval_tasks()
+
+    def start_message_dispenser(self):
+        """
+        Start the message dispenser.
 
-    def start_background_tasks(self):
         """
-        Start background tasks for the agent.
+        self._loop.create_task(self._dispenser.run())
+
+    def start_interval_tasks(self):
+        """
+        Start interval tasks for the agent.
 
         """
-        # Start the interval tasks
         for func, period in self._interval_handlers:
-            task = self._loop.create_task(_run_interval(func, self._ctx, period))
-            self._background_tasks.add(task)
-            task.add_done_callback(self._background_tasks.discard)
+            self._loop.create_task(_run_interval(func, self._ctx, period))
+
+    def start_message_receivers(self):
+        """
+        Start message receiving tasks for the agent.
 
+        """
         # start the background message queue processor
-        task = self._loop.create_task(self._process_message_queue())
-        self._background_tasks.add(task)
-        task.add_done_callback(self._background_tasks.discard)
+        self._loop.create_task(self._process_message_queue())
 
         # start the wallet messaging client if enabled
         if self._wallet_messaging_client is not None:
             for task in [
                 self._wallet_messaging_client.poll_server(),
                 self._wallet_messaging_client.process_message_queue(self._ctx),
             ]:
-                new_task = self._loop.create_task(task)
-                self._background_tasks.add(new_task)
-                new_task.add_done_callback(self._background_tasks.discard)
+                self._loop.create_task(task)
 
     def run(self):
         """
@@ -927,24 +1025,21 @@ async def _process_message_queue(self):
                 )
                 continue
 
-            context = Context(
-                self._identity.address,
-                self.identifier,
-                self._name,
-                self._storage,
-                self._resolver,
-                self._identity,
-                self._wallet,
-                self._ledger,
-                self._queries,
+            context = ExternalContext(
+                agent=self._ctx.agent,
+                storage=self._storage,
+                ledger=self._ledger,
+                resolver=self._resolver,
+                dispenser=self._dispenser,
+                wallet_messaging_client=self._wallet_messaging_client,
+                logger=self._logger,
+                queries=self._queries,
                 session=session,
                 replies=self._replies,
-                interval_messages=self._interval_messages,
                 message_received=MsgDigest(
                     message=message, schema_digest=schema_digest
                 ),
-                protocols=self.protocols,
-                logger=self._logger,
+                protocol=self.protocols.get(schema_digest),
             )
 
             # parse the received message

python/src/uagents/asgi.py

@@ -7,12 +7,13 @@
 import pydantic
 import uvicorn
 from requests.structures import CaseInsensitiveDict
-from uagents.config import RESPONSE_TIME_HINT_SECONDS, get_logger
+from uagents.config import RESPONSE_TIME_HINT_SECONDS
 from uagents.crypto import is_user_address
 from uagents.dispatch import dispatcher
 from uagents.envelope import Envelope
 from uagents.models import ErrorMessage
 from uagents.query import enclose_response_raw
+from uagents.utils import get_logger
 
 HOST = "0.0.0.0"