Merge pull request game-by-virtuals#110 from DPSN-org/feature/dpsn-plugin

Feature/dpsn plugin

Author: Ang-dot

plugins/dpsn/README.md

@@ -0,0 +1,204 @@
+# 🌐 DPSN Plugin for Virtuals Protocol (Python)
+
+> Decentralized Publish-Subscribe Network (DPSN) plugin for Virtuals Protocol agents, implemented in Python.
+
+[![Virtuals Protocol](https://img.shields.io/badge/Virtuals%20Protocol-plugin-blue)](https://virtuals.io/)
+[![Version](https://img.shields.io/badge/version-alpha-orange)](https://github.com/virtuals-protocol/virtuals-game-python)
+[![License](https://img.shields.io/badge/license-MIT-green)](../../LICENSE)
+
+## 📋 Overview
+
+This plugin enables Virtuals Protocol agents (written in Python) to connect to, subscribe to, and interact with data streams available on the [DPSN Data Streams Store](https://streams.dpsn.org/).
+
+Agents can leverage this plugin to consume real-time data for decision-making, reacting to events, or integrating external information feeds.
+
+To provide personalized data streams for your agents, you can create and publish data into your own DPSN topics using the [dpsn-client for Python](https://github.com/DPSN-org/dpsn-python-client).
+
+For more information, visit:
+-   [DPSN Official Website](https://dpsn.org)
+
+## ✨ Features
+
+-   **Seamless Integration**: Connects Virtuals Protocol agents (Python) to the DPSN decentralized pub/sub network.
+-   **Real-time Data Handling**: Subscribe to topics and process incoming messages via a configurable callback.
+-   **Topic Management**: Provides agent-executable functions to `subscribe` and `unsubscribe` from DPSN topics.
+-   **Error Handling**: Includes basic error handling and logging for connection and subscription issues.
+-   **Graceful Shutdown**: Allows the agent to explicitly shut down the DPSN connection.
+
+## ⚙️ Configuration
+
+Ensure the following environment variables are set, typically in a `.env` file in your project root:
+
+> **Note**: The EVM private key (`PVT_KEY`) is used solely for signing authentication messages with the DPSN network. This process does not execute any on-chain transactions or incur gas fees.
+
+```dotenv
+# Your EVM-compatible wallet private key (e.g., Metamask)
+PVT_KEY=your_evm_wallet_private_key_here
+
+# The URL of the DPSN node to connect to (e.g., betanet.dpsn.org)
+DPSN_URL=betanet.dpsn.org
+
+# Optional: Add VIRTUALS_API_KEY if required by your GameAgent setup
+# VIRTUALS_API_KEY=your_virtuals_api_key_here
+```
+
+## 📚 Usage
+
+### Basic Setup 
+
+The `DpsnPlugin` is designed to be used within the Virtuals Protocol Game SDK framework. You would typically instantiate it and potentially pass it to your `GameAgent` or similar construct.
+
+```python
+# Import the pre-instantiated plugin (recommended)
+from plugins.dpsn.dpsn_plugin_gamesdk.dpsn_plugin import DpsnPlugin
+load_dotenv()
+
+dpsn_plugin=DpsnPlugin(
+            dpsn_url=os.getenv("DPSN_URL"),
+            pvt_key=os.getenv("PVT_KEY")
+        )
+
+# Define a simple message handler
+def handle_message(message_data):
+    topic = message_data.get('topic', 'unknown')
+    payload = message_data.get('payload', {})
+    print(f"Message on {topic}: {payload}")
+
+# Register the message handler
+dpsn_plugin.set_message_callback(handle_message)
+
+# Subscribe to a topic
+status, message, details = dpsn_plugin.subscribe(
+    topic="0xe14768a6d8798e4390ec4cb8a4c991202c2115a5cd7a6c0a7ababcaf93b4d2d4/BTCUSDT/ticker"
+)
+print(f"Subscription status: {status}, Message: {message}")
+
+# Later when done:
+dpsn_plugin.unsubscribe(topic="0xe14768a6d8798e4390ec4cb8a4c991202c2115a5cd7a6c0a7ababcaf93b4d2d4/BTCUSDT/ticker")
+dpsn_plugin.shutdown()
+
+```
+
+### Interacting via Agent Tasks
+
+The Game Agent interacts with the plugin by executing tasks that map to the plugin's `Function` objects. The exact syntax depends on your Game SDK's `runTask` or equivalent method.
+
+```python
+from game_sdk.game.agent import Agent, WorkerConfig
+from game_sdk.game.custom_types import FunctionResult
+from dpsn_plugin_gamesdk.dpsn_plugin import DpsnPlugin
+
+load_dotenv()
+
+dpsn_plugin=DpsnPlugin(
+            dpsn_url=os.getenv("DPSN_URL"),
+            pvt_key=os.getenv("PVT_KEY")
+        )
+# --- Add Message Handler --- 
+def handle_incoming_message(message_data: dict):
+    """Callback function to process messages received via the plugin."""
+    try:
+        topic = message_data.get('topic', 'N/A')
+        payload = message_data.get('payload', '{}')
+        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+        print(f"\n--- Message Received ({timestamp}) ---")
+        print(f"Topic: {topic}")
+        # Pretty print payload if it's likely JSON/dict
+        if isinstance(payload, (dict, list)):
+            print(f"Payload:\n{json.dumps(payload, indent=2)}")
+            return payload
+        else:
+            print(f"Payload: {payload}")
+            return payload
+        print("-----------------------------------")
+    except Exception as e:
+        print(f"Error in message handler: {e}")
+
+# Set the callback in the plugin instance *before* running the agent
+dpsn_plugin.set_message_callback(handle_incoming_message)
+
+def get_agent_state_fn(function_result: FunctionResult, current_state: dict) -> dict:
+    """Update state based on the function results"""
+    init_state = {}
+
+    if current_state is None:
+        return init_state
+
+    if function_result.info is not None:
+        current_state.update(function_result.info)
+
+    return current_state
+
+def get_worker_state(function_result: FunctionResult, current_state: dict) -> dict:
+    """Update state based on the function results"""
+    init_state = {}
+
+    if current_state is None:
+        return init_state
+
+    if function_result.info is not None:
+        current_state.update(function_result.info)
+
+    return current_state
+
+
+subscription_worker = WorkerConfig(
+    id="subscription_worker",
+    worker_description="Worker specialized in managing DPSN topic subscriptions, unsubscriptions, message handling, and shutdown.",
+    get_state_fn=get_worker_state,
+    action_space=[
+        dpsn_plugin.get_function("subscribe"),
+        dpsn_plugin.get_function("unsubscribe"),
+        dpsn_plugin.get_function("shutdown")
+    ],
+)
+
+# Initialize the agent
+agent = Agent(
+    api_key=os.environ.get("GAME_API_KEY"),
+    name="DPSN Market Data Agent",
+    agent_goal="Monitor SOLUSDT market data from DPSN and process real-time updates.",
+    agent_description=(
+        "You are an AI agent specialized in DPSN market data processing"
+        "You can subscribe dpsn topic"
+        "after 5 minutes unsubscribe the topic"
+        "next 5 minutes close the connection"
+        "\n\nAvailable topics:"
+        "\n- 0xe14768a6d8798e4390ec4cb8a4c991202c2115a5cd7a6c0a7ababcaf93b4d2d4/SOLUSDT/ohlc"
+    ),
+    get_agent_state_fn=get_agent_state_fn,
+    workers=[
+        subscription_worker
+    ]
+)
+```
+
+## 📖 API Reference (`DpsnPlugin`)
+
+Key components of the `DpsnPlugin` class:
+
+-   `__init__(dpsn_url: Optional[str] = ..., pvt_key: Optional[str] = ...)`: Constructor. Reads credentials from env vars by default. Raises `ValueError` if credentials are missing.
+-   `get_function(fn_name: str) -> Function`: Retrieves the `Function` object (`subscribe`, `unsubscribe`, `shutdown`) for the Game SDK.
+-   `set_message_callback(callback: Callable[[Dict[str, Any]], None])`: Registers the function to call when a message is received on a subscribed topic. The callback receives a dictionary (structure depends on the underlying `dpsn-client`).
+-   `subscribe(topic: str) -> Tuple[FunctionResultStatus, str, Dict[str, Any]]`: (Executable Function) Subscribes to a topic. Handles initialization if needed. Returns status, message, and details.
+-   `unsubscribe(topic: str) -> Tuple[FunctionResultStatus, str, Dict[str, Any]]`: (Executable Function) Unsubscribes from a topic. Handles initialization if needed. Returns status, message, and details.
+-   `shutdown() -> Tuple[FunctionResultStatus, str, Dict[str, Any]]`: (Executable Function) Disconnects the DPSN client gracefully. Returns status, message, and details.
+-   `_ensure_initialized()`: (Internal) Manages the lazy initialization of the DPSN client connection. Raises `DpsnInitializationError` on failure.
+
+### Agent-Executable Functions
+
+The plugin exposes the following functions intended to be called via the Game Agent's task execution mechanism:
+
+-   **`subscribe`**:
+    -   Description: Subscribe to a DPSN topic.
+    -   Args: `topic` (string, required) - The topic to subscribe to.
+-   **`unsubscribe`**:
+    -   Description: Unsubscribe from a DPSN topic.
+    -   Args: `topic` (string, required) - The topic to unsubscribe from.
+-   **`shutdown`**:
+    -   Description: Shutdown the DPSN client connection.
+    -   Args: None.
+
+
+
+> In case of any queries regarding DPSN, please reach out to the team on [Telegram](https://t.me/dpsn_dev) 📥.