Merge pull request #97 from niceberginc/use_tensor

use_tensor

Author: tkorkmazeth

agentipy/agent/__init__.py

@@ -2750,4 +2750,46 @@ def simulate_switchboard_feed(
             from agentipy.tools.use_switchboard import SwitchboardManager
             return SwitchboardManager.simulate_switchboard_feed(self, feed, crossbar_url)
         except Exception as e:
-            raise SolanaAgentKitError(f"Failed to simulate Switchboard feed: {e}")
+            raise SolanaAgentKitError(f"Failed to simulate Switchboard feed: {e}")
+    
+    def list_nft_for_sale(
+        self,
+        price: float,
+        nft_mint: str,
+    ) -> Optional[Dict[str, Any]]:
+        """
+        List an NFT for sale.
+
+        Args:
+            agent (SolanaAgentKit): The Solana agent instance.
+            price (float): The sale price.
+            nft_mint (str): The NFT mint address.
+
+        Returns:
+            dict: Transaction details.
+        """
+        try:
+            from agentipy.tools.use_tensor import TensorManager
+            return TensorManager.list_nft_for_sale(self, price, nft_mint)
+        except Exception as e:
+            raise SolanaAgentKitError(f"Failed to list NFT for sale: {e}")
+        
+    def cancel_listing(
+        self,
+        nft_mint: str,
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Cancel an NFT listing.
+
+        Args:
+            agent (SolanaAgentKit): The Solana agent instance.
+            nft_mint (str): The NFT mint address.
+
+        Returns:
+            dict: Transaction details.
+        """
+        try:
+            from agentipy.tools.use_tensor import TensorManager
+            return TensorManager.cancel_listing(self, nft_mint)
+        except Exception as e:
+            raise SolanaAgentKitError(f"Failed to cancel listing: {e}")

agentipy/langchain/__init__.py

@@ -34,6 +34,7 @@
 from agentipy.langchain.squads import get_squads_tools
 from agentipy.langchain.stork import get_stork_tools
 from agentipy.langchain.switchboard import get_switchboard_tools
+from agentipy.langchain.tensor import get_tensor_tools
 
 
 def create_solana_tools(solana_kit: SolanaAgentKit):
@@ -77,4 +78,5 @@ def create_solana_tools(solana_kit: SolanaAgentKit):
         *get_solutiofi_tools(solana_kit=solana_kit),
         *get_squads_tools(solana_kit=solana_kit),
         *get_switchboard_tools(solana_kit=solana_kit),
+        *get_tensor_tools(solana_kit=solana_kit),
     ]   

agentipy/langchain/tensor/__init__.py

@@ -0,0 +1,10 @@
+from agentipy.agent import SolanaAgentKit
+from agentipy.langchain.tensor.trade import (TensorCancelListingTool,
+                                             TensorListNFTForSaleTool)
+
+
+def get_tensor_tools(solana_kit: SolanaAgentKit):
+    return [
+        TensorCancelListingTool(solana_kit),
+        TensorListNFTForSaleTool(solana_kit),
+    ]

agentipy/langchain/tensor/trade.py

@@ -0,0 +1,95 @@
+import json
+
+from langchain.tools import BaseTool
+
+from agentipy.agent import SolanaAgentKit
+from agentipy.helpers import validate_input
+
+
+class TensorListNFTForSaleTool(BaseTool):
+    name: str = "tensor_list_nft_for_sale"
+    description: str = """
+    Lists an NFT for sale on Tensor.
+
+    Input: A JSON string with:
+    {
+        "price": "float, the sale price of the NFT",
+        "nft_mint": "string, the NFT mint address"
+    }
+
+    Output:
+    {
+        "transaction_details": "dict, transaction details",
+        "message": "string, if an error occurs"
+    }
+    """
+    solana_kit: SolanaAgentKit
+
+    async def _arun(self, input: str):
+        try:
+            data = json.loads(input)
+            schema = {
+                "price": {"type": float, "required": True},
+                "nft_mint": {"type": str, "required": True},
+            }
+            validate_input(data, schema)
+
+            price = data["price"]
+            nft_mint = data["nft_mint"]
+
+            result = await self.solana_kit.list_nft_for_sale(price, nft_mint)
+            return {
+                "status": "success",
+                "transaction_details": result,
+            }
+        except Exception as e:
+            return {
+                "status": "error",
+                "message": str(e),
+            }
+
+    def _run(self, input: str):
+        raise NotImplementedError("This tool only supports async execution via _arun. Please use the async interface.")
+
+
+class TensorCancelListingTool(BaseTool):
+    name: str = "tensor_cancel_listing"
+    description: str = """
+    Cancels an NFT listing on Tensor.
+
+    Input: A JSON string with:
+    {
+        "nft_mint": "string, the NFT mint address"
+    }
+
+    Output:
+    {
+        "transaction_details": "dict, transaction details",
+        "message": "string, if an error occurs"
+    }
+    """
+    solana_kit: SolanaAgentKit
+
+    async def _arun(self, input: str):
+        try:
+            data = json.loads(input)
+            schema = {
+                "nft_mint": {"type": str, "required": True},
+            }
+            validate_input(data, schema)
+
+            nft_mint = data["nft_mint"]
+
+            result = await self.solana_kit.cancel_listing(nft_mint)
+            return {
+                "status": "success",
+                "transaction_details": result,
+            }
+        except Exception as e:
+            return {
+                "status": "error",
+                "message": str(e),
+            }
+
+    def _run(self, input: str):
+        raise NotImplementedError("This tool only supports async execution via _arun. Please use the async interface.")

agentipy/tools/use_tensor.py

@@ -0,0 +1,104 @@
+import logging
+from typing import Any, Dict, Optional
+
+import requests
+
+from agentipy.agent import SolanaAgentKit
+from agentipy.utils.agentipy_proxy.utils import encrypt_private_key
+
+logger = logging.getLogger(__name__)
+
+
+class TensorManager:
+    """
+    Manager class to handle Tensor functionalities such as listing NFTs for sale and canceling listings.
+    """
+
+    @staticmethod
+    def list_nft_for_sale(
+        agent: SolanaAgentKit,
+        price: float,
+        nft_mint: str,
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Lists an NFT for sale on Tensor.
+
+        :param agent: An instance of SolanaAgentKit.
+        :param price: The listing price of the NFT.
+        :param nft_mint: The mint address of the NFT.
+        :return: Transaction signature or error details.
+        """
+        try:
+            if not price or not nft_mint:
+                raise ValueError("Price and NFT mint address are required.")
+
+            encrypted_private_key = encrypt_private_key(agent.private_key)
+
+            payload = {
+                "requestId": encrypted_private_key["requestId"],
+                "encrypted_private_key": encrypted_private_key["encryptedPrivateKey"],
+                "rpc_url": agent.rpc_url,
+                "open_api_key": agent.openai_api_key,
+                "price": price,
+                "nftMint": nft_mint,
+            }
+
+            response = requests.post(
+                f"{agent.base_proxy_url}/{agent.api_version}/tensor/list-nft-for-sale",
+                json=payload,
+                headers={"Content-Type": "application/json"},
+            )
+            response.raise_for_status()
+
+            data = response.json()
+            return data if data.get("success") else {"success": False, "error": data.get("error", "Unknown error")}
+
+        except requests.exceptions.RequestException as http_error:
+            logger.error(f"HTTP error during NFT listing: {http_error}", exc_info=True)
+            return {"success": False, "error": str(http_error)}
+        except Exception as error:
+            logger.error(f"Unexpected error during NFT listing: {error}", exc_info=True)
+            return {"success": False, "error": str(error)}
+
+    @staticmethod
+    def cancel_listing(
+        agent: SolanaAgentKit,
+        nft_mint: str,
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Cancels a listed NFT on Tensor.
+
+        :param agent: An instance of SolanaAgentKit.
+        :param nft_mint: The mint address of the NFT to cancel the listing.
+        :return: Transaction signature or error details.
+        """
+        try:
+            if not nft_mint:
+                raise ValueError("NFT mint address is required.")
+
+            encrypted_private_key = encrypt_private_key(agent.private_key)
+
+            payload = {
+                "requestId": encrypted_private_key["requestId"],
+                "encrypted_private_key": encrypted_private_key["encryptedPrivateKey"],
+                "rpc_url": agent.rpc_url,
+                "open_api_key": agent.openai_api_key,
+                "nftMint": nft_mint,
+            }
+
+            response = requests.post(
+                f"{agent.base_proxy_url}/{agent.api_version}/tensor/cancel-listing",
+                json=payload,
+                headers={"Content-Type": "application/json"},
+            )
+            response.raise_for_status()
+
+            data = response.json()
+            return data if data.get("success") else {"success": False, "error": data.get("error", "Unknown error")}
+
+        except requests.exceptions.RequestException as http_error:
+            logger.error(f"HTTP error during listing cancellation: {http_error}", exc_info=True)
+            return {"success": False, "error": str(http_error)}
+        except Exception as error:
+            logger.error(f"Unexpected error during listing cancellation: {error}", exc_info=True)
+            return {"success": False, "error": str(error)}