feat: Add Python implementation of rugcheck plugin (#204)

* feat: Add Python implementation of rugcheck plugin

- Implement RugCheck API integration with JWT authentication
- Add all endpoints from TypeScript implementation
- Follow coingecko plugin Python structure
- Include proper parameter definitions
- Set up project configuration

* chore: address PR feedback - remove empty init file, add README.md, fix pyproject.toml packages config

Co-Authored-By: Agus Armellini Fischer <[email protected]>

* Fix PR

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Agus Armellini Fischer <[email protected]>
Co-authored-by: Agustin Armellini Fischer <[email protected]>

Author: 3 people

python/examples/langchain/web3/example.py

@@ -5,8 +5,8 @@
 load_dotenv()
 
 from langchain_openai import ChatOpenAI
-from langchain.agents import AgentExecutor, create_structured_chat_agent
-from langchain.hub import pull
+from langchain.agents import AgentExecutor, create_tool_calling_agent
+from langchain_core.prompts import ChatPromptTemplate
 from web3 import Web3
 from web3.middleware.signing import construct_sign_and_send_raw_middleware
 from eth_account.signers.local import LocalAccount
@@ -17,6 +17,8 @@
 from goat_plugins.erc20 import erc20, ERC20PluginOptions
 from goat_wallets.evm import send_eth
 from goat_wallets.web3 import Web3EVMWalletClient
+from goat_plugins.coingecko import coingecko, CoinGeckoPluginOptions
+from goat_plugins.rugcheck import rugcheck
 
 # Initialize Web3 and account
 w3 = Web3(Web3.HTTPProvider(os.getenv("RPC_PROVIDER_URL")))
@@ -36,24 +38,43 @@
 
 def main():
     # Get the prompt template
-    prompt = pull("hwchase17/structured-chat-agent")
+    prompt = ChatPromptTemplate.from_messages(
+    [
+        ("system", "You are a helpful assistant"),
+        ("placeholder", "{chat_history}"),
+        ("human", "{input}"),
+        ("placeholder", "{agent_scratchpad}"),
+    ]
+)
 
     # Initialize tools with web3 wallet
     tools = get_on_chain_tools(
         wallet=Web3EVMWalletClient(w3),
-        plugins=[send_eth(), erc20(options=ERC20PluginOptions(tokens=[USDC, PEPE]))],
+        plugins=[
+            send_eth(),
+            erc20(options=ERC20PluginOptions(tokens=[USDC, PEPE])),
+            coingecko(options=CoinGeckoPluginOptions(api_key=os.getenv("COINGECKO_API_KEY")))
+        ],
     )
+    
+    agent = create_tool_calling_agent(llm, tools, prompt)
+    agent_executor = AgentExecutor(agent=agent, tools=tools, handle_parsing_errors=True, verbose=True)
+    
+    while True:
+        user_input = input("\nYou: ").strip()
+        
+        if user_input.lower() == 'quit':
+            print("Goodbye!")
+            break
+            
+        try:
+            response = agent_executor.invoke({
+                "input": user_input,
+            })
 
-    # Create the agent
-    agent = create_structured_chat_agent(llm=llm, tools=tools, prompt=prompt)
-
-    # Create the executor
-    agent_executor = AgentExecutor(agent=agent, tools=tools, handle_parsing_errors=True)
-
-    # Execute the agent
-    response = agent_executor.invoke({"input": "Get my balance in USDC"})
-
-    print(response)
+            print("\nAssistant:", response["output"])
+        except Exception as e:
+            print("\nError:", str(e))
 
 
 if __name__ == "__main__":

python/examples/langchain/web3/poetry.lock

@@ -8,7 +8,7 @@ keywords = ["goat", "sdk", "web3", "agents", "ai"]
 homepage = "https://ohmygoat.dev/"
 repository = "https://github.com/goat-sdk/goat"
 packages = [
-    { include = "src" },
+    { include = "example.py" },
 ]
 
 [tool.poetry.dependencies]
@@ -22,6 +22,7 @@ goat-sdk-wallet-evm = "^0.1.0"
 goat-sdk-wallet-web3 = "^0.1.0"
 goat-sdk-plugin-erc20 = "^0.1.0"
 goat-sdk-adapter-langchain = "^0.1.0"
+goat-sdk-plugin-coingecko = "^0.1.0"
 
 [tool.poetry.group.test.dependencies]
 pytest = "^8.3.4"
@@ -47,6 +48,7 @@ goat-sdk = { path = "../../../src/goat-sdk", develop = true }
 goat-sdk-wallet-evm = { path = "../../../src/wallets/evm", develop = true }
 goat-sdk-wallet-web3 = { path = "../../../src/wallets/web3", develop = true }
 goat-sdk-plugin-erc20 = { path = "../../../src/plugins/erc20", develop = true }
+goat-sdk-plugin-coingecko = { path = "../../../src/plugins/coingecko", develop = true }
 goat-sdk-adapter-langchain = { path = "../../../src/adapters/langchain", develop = true }
 
 [tool.ruff]

python/examples/langchain/web3/pyproject.toml

@@ -17,7 +17,7 @@ async def get_trending_coins(self, parameters: dict):
             url = f"{self.base_url}/search/trending?x_cg_demo_api_key={self.api_key}"
             async with session.get(url) as response:
                 if not response.ok:
-                    raise Exception(f"HTTP error! status: {response.status}")
+                    raise Exception(f"HTTP error! status: {response.status} {await response.text()}")
                 return await response.json()
 
     @Tool({
@@ -40,7 +40,7 @@ async def get_coin_price(self, parameters: dict):
             url = f"{self.base_url}/simple/price"
             async with session.get(url, params=params) as response:
                 if not response.ok:
-                    raise Exception(f"HTTP error! status: {response.status}")
+                    raise Exception(f"HTTP error! status: {response.status} {await response.text()}")
                 return await response.json()
 
     @Tool({
@@ -58,7 +58,7 @@ async def search_coins(self, parameters: dict):
             url = f"{self.base_url}/search"
             async with session.get(url, params=params) as response:
                 if not response.ok:
-                    raise Exception(f"HTTP error! status: {response.status}")
+                    raise Exception(f"HTTP error! status: {response.status} {await response.text()}")
                 data = await response.json()
 
                 if parameters["exact_match"]:

python/src/plugins/coingecko/goat_plugins/coingecko/service.py

@@ -0,0 +1,57 @@
+# GOAT SDK RugCheck Plugin
+
+A Python implementation of the RugCheck plugin for the GOAT SDK, providing access to RugCheck's token analysis and verification services.
+
+## Features
+
+- Get recently detected tokens
+- Get trending tokens in the last 24h
+- Get most voted tokens in the last 24h
+- Get recently verified tokens
+- Generate token report summaries
+
+## Installation
+
+```bash
+poetry install
+```
+
+## Usage
+
+```python
+from goat_plugins.rugcheck import rugcheck, RugCheckPluginOptions
+
+# Initialize the plugin
+options = RugCheckPluginOptions(jwt_token="your_jwt_token")  # JWT token is optional
+plugin = rugcheck(options)
+
+# Example: Get recently detected tokens
+async def get_recent_tokens():
+    service = plugin.services[0]
+    tokens = await service.get_recently_detected_tokens({})
+    return tokens
+
+# Example: Generate a token report
+async def get_token_report(mint_address: str):
+    service = plugin.services[0]
+    report = await service.generate_token_report_summary({"mint": mint_address})
+    return report
+```
+
+## Authentication
+
+The plugin supports JWT token authentication. While optional, some endpoints may require authentication for full access.
+
+## Rate Limiting
+
+The plugin includes built-in handling for rate limits (HTTP 429 responses) from the RugCheck API.
+
+## Development
+
+- Python 3.9+
+- Uses `aiohttp` for async HTTP requests
+- Uses `pydantic` for parameter validation
+
+## License
+
+See the main GOAT SDK repository for license information.

python/src/plugins/rugcheck/README.md

@@ -0,0 +1,21 @@
+from dataclasses import dataclass
+
+from goat.classes.plugin_base import PluginBase
+from .service import RugCheckService
+
+
+@dataclass
+class RugCheckPluginOptions:
+    jwt_token: str = ""
+
+
+class RugCheckPlugin(PluginBase):
+    def __init__(self):
+        super().__init__("rugcheck", [RugCheckService()])
+
+    def supports_chain(self, chain) -> bool:
+        return True
+
+
+def rugcheck() -> RugCheckPlugin:
+    return RugCheckPlugin()

python/src/plugins/rugcheck/goat_plugins/rugcheck/__init__.py

@@ -0,0 +1,11 @@
+from pydantic import BaseModel, Field
+
+
+class GetTokenReportParameters(BaseModel):
+    mint: str = Field(
+        description="The token mint address to generate the report for"
+    )
+
+
+class NoParameters(BaseModel):
+    pass

python/src/plugins/rugcheck/goat_plugins/rugcheck/parameters.py

@@ -0,0 +1,64 @@
+import aiohttp
+from goat.decorators.tool import Tool
+from .parameters import GetTokenReportParameters, NoParameters
+
+
+class RugCheckService:
+    def __init__(self, jwt_token: str = ""):
+        self.jwt_token = jwt_token
+        self.base_url = "https://api.rugcheck.xyz/v1"
+
+    async def _make_request(self, endpoint: str):
+        headers = {
+            "Content-Type": "application/json",
+        }
+        
+        async with aiohttp.ClientSession() as session:
+            url = f"{self.base_url}{endpoint}"
+            async with session.get(url, headers=headers) as response:
+                if not response.ok:
+                    if response.status == 429:
+                        raise Exception("RugCheck API rate limit exceeded")
+                    raise Exception(f"RugCheck API request failed: {response.status}")
+                return await response.json()
+
+    @Tool({
+        "description": "Get recently detected tokens from RugCheck",
+        "parameters_schema": NoParameters
+    })
+    async def get_recently_detected_tokens(self, parameters: dict):
+        """Get recently detected tokens from RugCheck"""
+        return await self._make_request("/stats/new_tokens")
+
+    @Tool({
+        "description": "Get trending tokens in the last 24h from RugCheck",
+        "parameters_schema": NoParameters
+    })
+    async def get_trending_tokens_24h(self, parameters: dict):
+        """Get trending tokens in the last 24h from RugCheck"""
+        return await self._make_request("/stats/trending")
+
+    @Tool({
+        "description": "Get tokens with the most votes in the last 24h from RugCheck",
+        "parameters_schema": NoParameters
+    })
+    async def get_most_voted_tokens_24h(self, parameters: dict):
+        """Get tokens with the most votes in the last 24h from RugCheck"""
+        return await self._make_request("/stats/recent")
+
+    @Tool({
+        "description": "Get recently verified tokens from RugCheck",
+        "parameters_schema": NoParameters
+    })
+    async def get_recently_verified_tokens(self, parameters: dict):
+        """Get recently verified tokens from RugCheck"""
+        return await self._make_request("/stats/verified")
+
+    @Tool({
+        "description": "Generate a report summary for the given token mint",
+        "parameters_schema": GetTokenReportParameters
+    })
+    async def generate_token_report_summary(self, parameters: dict):
+        """Generate a report summary for the given token mint"""
+        mint = parameters["mint"]
+        return await self._make_request(f"/tokens/{mint}/report/summary")