Merge pull request #96 from niceberginc/use_switchboard

use_swithcboard

Author: tkorkmazeth

agentipy/agent/__init__.py

@@ -2728,4 +2728,26 @@ def transfer_from_multisig_treasury(
             from agentipy.tools.use_squads import SquadsManager
             return SquadsManager.transfer_from_multisig_treasury(self, amount, to, vault_index, mint)
         except Exception as e:
-            raise SolanaAgentKitError(f"Failed to transfer from multisig treasury: {e}")
+            raise SolanaAgentKitError(f"Failed to transfer from multisig treasury: {e}")
+        
+    def simulate_switchboard_feed(
+        self,
+        feed: str,
+        crossbar_url: Optional[str] = None,
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Simulate a Switchboard feed.
+
+        Args:
+            agent (SolanaAgentKit): The Solana agent instance.
+            feed (str): The feed name.
+            crossbar_url (str): The Crossbar URL.
+
+        Returns:
+            dict: Simulation details.
+        """
+        try:
+            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}")

agentipy/langchain/__init__.py

@@ -33,6 +33,7 @@
 from agentipy.langchain.solutiofi import get_solutiofi_tools
 from agentipy.langchain.squads import get_squads_tools
 from agentipy.langchain.stork import get_stork_tools
+from agentipy.langchain.switchboard import get_switchboard_tools
 
 
 def create_solana_tools(solana_kit: SolanaAgentKit):
@@ -75,4 +76,5 @@ def create_solana_tools(solana_kit: SolanaAgentKit):
         *get_allora_tools(solana_kit=solana_kit),
         *get_solutiofi_tools(solana_kit=solana_kit),
         *get_squads_tools(solana_kit=solana_kit),
+        *get_switchboard_tools(solana_kit=solana_kit),
     ]   

agentipy/langchain/switchboard/__init__.py

@@ -0,0 +1,9 @@
+from agentipy.agent import SolanaAgentKit
+from agentipy.langchain.switchboard.simulate_switchboard_feed import \
+    SwitchboardSimulateFeedTool
+
+
+def get_switchboard_tools(solana_kit: SolanaAgentKit):
+    return [
+        SwitchboardSimulateFeedTool(solana_kit=solana_kit)
+    ]

agentipy/langchain/switchboard/simulate_switchboard_feed.py

@@ -0,0 +1,52 @@
+import json
+
+from langchain.tools import BaseTool
+
+from agentipy.agent import SolanaAgentKit
+from agentipy.helpers import validate_input
+
+
+class SwitchboardSimulateFeedTool(BaseTool):
+    name: str = "switchboard_simulate_feed"
+    description: str = """
+    Simulate a Switchboard feed.
+
+    Input: A JSON string with:
+    {
+        "feed": "string, the name of the Switchboard feed to simulate.",
+        "crossbar_url": "string, optional, the Crossbar URL for additional configuration."
+    }
+
+    Output:
+    {
+        "simulation_details": "dict, the details of the simulation",
+        "message": "string, if an error occurs"
+    }
+    """
+    solana_kit: SolanaAgentKit
+
+    async def _arun(self, input: str):
+        try:
+            data = json.loads(input)
+            schema = {
+                "feed": {"type": str, "required": True},
+                "crossbar_url": {"type": str, "required": False},
+            }
+            validate_input(data, schema)
+
+            feed = data["feed"]
+            crossbar_url = data.get("crossbar_url")
+
+            result = await self.solana_kit.simulate_switchboard_feed(feed, crossbar_url)
+            return {
+                "status": "success",
+                "simulation_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_switchboard.py

@@ -0,0 +1,61 @@
+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 SwitchboardManager:
+    """
+    Manager class to handle Switchboard functionalities such as simulating a Switchboard feed.
+    """
+
+    @staticmethod
+    def simulate_switchboard_feed(
+        agent: SolanaAgentKit,
+        feed: str,
+        crossbar_url: Optional[str] = None,
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Simulates a Switchboard feed.
+
+        :param agent: An instance of SolanaAgentKit.
+        :param feed: The feed to be simulated.
+        :param crossbar_url: The crossbar URL for Switchboard (default provided).
+        :return: The result of the simulation or error details.
+        """
+        try:
+            if not feed:
+                raise ValueError("Feed 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,
+                "feed": feed,
+                "crossbarUrl": crossbar_url if crossbar_url else None
+            }
+
+            response = requests.post(
+                f"{agent.base_proxy_url}/{agent.api_version}/switchboard/simulate-switchboard-feed",
+                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 Switchboard feed simulation: {http_error}", exc_info=True)
+            return {"success": False, "error": str(http_error)}
+        except Exception as error:
+            logger.error(f"Unexpected error during Switchboard feed simulation: {error}", exc_info=True)
+            return {"success": False, "error": str(error)}