rpc and edits

.chainlit/config.toml

@@ -54,13 +54,13 @@ auto_tag_thread = true
 
 [UI]
 # Name of the app and chatbot.
-name = "Chatxbt"
+name = "ChatXBT"
 
 # Show the readme while the thread is empty.
 show_readme_as_default = true
 
 # Description of the app and chatbot. This is used for HTML tags.
-# description = ""
+# description = "AI-powered unification and execution protocol assitant for web3"
 
 # Large size content are by default collapsed for a cleaner ui
 default_collapse_content = true

chatxbt-assistant.py

@@ -15,7 +15,7 @@ from phi.storage.assistant.postgres import PgAssistantStorage
 from src.knowledge_bases.combined import knowledge_base
 
 storage = PgAssistantStorage(
-    # store runs in the ai.assistant_runs table
+    # stores runs in the ai.assistant_runs table
     table_name="assistant_runs",
     db_engine=sqlalchemy_engine
 )
@@ -57,15 +57,19 @@ async def set_starters():
             )
         ]
 
-
 @cl.on_chat_start
-def start():
+async def start():
     is_dev_mode = True if os.getenv("DEV_MODE") else False
 
     # Initialize the assistant
     cxbt_assistant = Assistant(
         llm=OpenAIChat(model="gpt-4o"),
-        tools=[CryptoDataTools(), CryptoSwapTools(), DuckDuckGo(), YFinanceTools(stock_price=True)],
+        tools=[
+            DuckDuckGo(),
+            CryptoDataTools(),
+            CryptoSwapTools(),
+            YFinanceTools(stock_price=True)
+        ],
         show_tool_calls= is_dev_mode,
         markdown=True,
         knowledge_base=knowledge_base,
@@ -82,11 +86,11 @@ def start():
 
 @cl.on_message
 async def main(message: cl.Message):
+
     # Retrieve the assistant from the user session
     agent = cl.user_session.get("agent")
 
     # Process the user message using the assistant
-    # response = agent.run(message.content, stream=True)
     response = ""
     for delta in agent.run(message.content, stream=True):
         response += delta

run.py

@@ -1,14 +1,35 @@
 import json
 import pprint
+import httpx
 import asyncio
 
-# import src.knowledge_bases.pipeline as Pipeline
+# # import src.knowledge_bases.pipeline as Pipeline
 
-# asyncio.run(Pipeline.main())
+# # asyncio.run(Pipeline.main())
 
-from src.data_sources.coin_gecko import CoinGecko
-from src.data_sources.cryptocompare import CryptoCompare
+# from src.data_sources.coin_gecko import CoinGecko
+# from src.data_sources.cryptocompare import CryptoCompare
 
-cgc = CoinGecko()
+# cgc = CoinGecko()
 
-pprint.pprint(cgc.get_trending_coin_list())
+# pprint.pprint(cgc.get_trending_coin_list())
+
+
+from src.libs.rpc_client import rpc_call
+
+# response = rpc_call(method_name="getSwapSources")
+
+async def main():
+    method_name = 'getSwapSources'
+
+    try:
+        response = await rpc_call(method_name)
+        if response is not None:
+            print("RPC call successful!")
+            print(response)
+        else:
+            print("RPC call failed.")
+    except httpx.RequestError as e:
+        print(f"Error in main function: {e}")
+
+asyncio.run(main())

src/libs/rpc_client.py

@@ -1,14 +1,29 @@
-import os
+import httpx, uuid, os
+from typing import Optional, Union
 from dotenv import load_dotenv
-
-import httpx
-from rpcpy.client import Client
-from rpcpy.serializers import MsgpackSerializer
+from src.libs.logger import logger
 
 load_dotenv()
 
-rpc_client = Client(
-    httpx.AsyncClient(),
-    request_serializer=MsgpackSerializer(),
-    base_url=os.getenv("CHATXBT_RPC_SERVER_URL")
-)
+rpc_server_url = os.getenv('CHATXBT_RPC_SERVER_URL')
+
+async def rpc_call(method_name, params: Optional[Union[dict, list]] = {}, url: str = rpc_server_url):
+    headers = {
+        'Content-Type': 'application/json',
+    }
+
+    payload = {
+        'method': method_name,
+        'params': params,
+        'jsonrpc': '2.0',
+        'id': str(uuid.uuid4()),
+    }
+
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(url, json=payload, headers=headers)
+            response.raise_for_status()
+            return response.json()
+    except httpx.RequestError as e:
+        print(f"Error making RPC call: {e}")
+        return None

src/tools/crypto_swap_toolkit.py

@@ -1,4 +1,6 @@
+import asyncio
 import json
+import pprint
 import requests
 from web3 import Web3
 from phi.tools import Toolkit
@@ -6,6 +8,7 @@ from phi.utils.log import logger
 from src.libs.web3 import get_web3_instance
 from src.libs.helper_functions import get_headers, get_private_key
 from src.libs.token_approval_helper import TokenApprovalHelper
+from src.libs.rpc_client import rpc_call
 
 class CryptoSwapTools(Toolkit):
     def __init__(self, web3: Web3 = get_web3_instance()):
@@ -45,10 +48,8 @@ class CryptoSwapTools(Toolkit):
                 'sellToken': sell_token,
                 'sellAmount': sell_amount
             }
-            response = requests.get("https://api.0x.org/swap/v1/quote", params=params, headers=get_headers())
-            response.raise_for_status()
-            # return response.json()
-            return f"{(response.json())}"
+            response = asyncio.run(rpc_call(method_name="getSwapQuote", params=params))
+            return f"{response}"
         except requests.exceptions.RequestException as e:
             logger.warning(f"Failed to get swap quote: {e}")
             # return {"error": str(e)}
@@ -76,10 +77,8 @@ class CryptoSwapTools(Toolkit):
                 'sellToken': sell_token,
                 'buyAmount': buy_amount
             }
-            response = requests.get("https://api.0x.org/swap/v1/price", params=params, headers=get_headers())
-            response.raise_for_status()
-            # return response.json()
-            return f"{(response.json())}"
+            response = asyncio.run(rpc_call(method_name="getSwapPrice", params=params))
+            return f"{response}"
         except requests.exceptions.RequestException as e:
             logger.warning(f"Failed to get swap price: {e}")
             # return {"error": str(e)}
@@ -96,15 +95,9 @@ class CryptoSwapTools(Toolkit):
         >>> get_swap_sources()
         """
         logger.info("Fetching swap sources")
-        try:
-            response = requests.get("https://api.0x.org/swap/v1/sources", headers=get_headers())
-            response.raise_for_status()
-            # return response.json()
-            return f"{(response.json())}"
-        except requests.exceptions.RequestException as e:
-            logger.warning(f"Failed to get swap sources: {e}")
-            # return {"error": str(e)}
-            return f"Error: {e}"
+
+        response = asyncio.run(rpc_call(method_name="getSwapSources"))
+        return f"{response}"
 
     def execute_swap(self, buy_token: str, sell_token: str, sell_amount: str, eth_address: str) -> str:
         """
@@ -161,22 +154,3 @@ class CryptoSwapTools(Toolkit):
             logger.warning(f"Failed to execute swap: {e}")
             # return {"error": str(e)}
             return f"Error: {e}"
-
-
-# Example usage
-if __name__ == "__main__":
-    eth_address = '0xYourEthereumAddress'  # Replace with your Ethereum address
-
-    # Initialize Web3
-    web3 = Web3(Web3.HTTPProvider('YOUR_INFURA_OR_ALCHEMY_URL'))
-
-    # Initialize CryptoSwapTools
-    swap_tool = CryptoSwapTools(web3)
-
-    # Fetch and print swap sources
-    sources = swap_tool.get_swap_sources()
-    print(f"Swap Sources: {sources}")
-
-    # Execute a swap (example)
-    receipt = swap_tool.execute_swap('DAI', 'ETH', '1000000000000000000', eth_address)
-    print(f"Swap Transaction Receipt: {receipt}")