shoko/randebu
1
0
Fork 0
Code Issues 6 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor: use subprocess to call ave-cloud-skill CLI scripts

Browse Source 
Instead of importing library functions directly, now calling the
ave_data_rest.py CLI script via subprocess. This follows the
recommended approach from the SKILL.md documentation.

Changes:
- Add _call_ave_script helper method for subprocess calls
- Update search_tokens, get_token, get_price, get_risk to use CLI
- Set AVE_USE_DOCKER=false to run scripts directly without Docker
- Remove direct imports of ave.http module
This commit is contained in:
shokollm
2026-04-13 10:24:42 +00:00
parent 99dded8d16
commit 2b7f54703e
1 changed files with 214 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files

240
src/backend/app/services/ai_agent/conversational.py
View File

@@ -317,29 +317,20 @@ class ConversationalAgent:
keyword = args.get("keyword", "") keyword = args.get("keyword", "")
limit = args.get("limit", 10) limit = args.get("limit", 10)
# Execute the tool using ave-cloud-skill code, output = self._call_ave_script(
import os "search",
import asyncio [
from ...core.config import get_settings "--keyword",
keyword,
settings = get_settings() "--chain",
os.environ["AVE_API_KEY"] = settings.AVE_API_KEY "bsc",
os.environ["API_PLAN"] = settings.AVE_API_PLAN "--limit",
str(limit),
from ave.http import api_get ],
resp = asyncio.run(
api_get(
"/tokens",
{
"keyword": keyword,
"limit": limit,
"chain": "bsc",
},
) )
) if code == 0:
if resp.status_code == 200: try:
data = resp.json() data = json.loads(output)
tokens = data.get("data", {}).get("tokens", []) tokens = data.get("data", {}).get("tokens", [])
if tokens: if tokens:
token_list = "" token_list = ""
@@ -355,10 +346,12 @@ class ConversationalAgent:
response_text = f"Here are the search results for '{keyword}' on BSC:\n\n{token_list}\nWould you like me to set up a strategy for any of these?" response_text = f"Here are the search results for '{keyword}' on BSC:\n\n{token_list}\nWould you like me to set up a strategy for any of these?"
else: else:
response_text = f"No tokens found for '{keyword}'. Try a different keyword." response_text = f"No tokens found for '{keyword}'. Try a different keyword."
else: except json.JSONDecodeError:
response_text = ( response_text = (
f"Failed to search tokens: {resp.status_code}" f"Failed to parse search results."
) )
else:
response_text = f"Failed to search tokens: {output}"
return { return {
"response": response_text, "response": response_text,
@@ -372,21 +365,12 @@ class ConversationalAgent:
address = args.get("address", "") address = args.get("address", "")
chain = args.get("chain", "bsc") chain = args.get("chain", "bsc")
import os code, output = self._call_ave_script(
import asyncio "token", ["--address", address, "--chain", chain]
from ...core.config import get_settings
settings = get_settings()
os.environ["AVE_API_KEY"] = settings.AVE_API_KEY
os.environ["API_PLAN"] = settings.AVE_API_PLAN
from ave.http import api_get
resp = asyncio.run(
api_get(f"/tokens/{address}-{chain}")
) )
if resp.status_code == 200: if code == 0:
data = resp.json() try:
data = json.loads(output)
token_data = data.get("data", {}) token_data = data.get("data", {})
if token_data: if token_data:
symbol = token_data.get("symbol", "N/A") symbol = token_data.get("symbol", "N/A")
@@ -402,9 +386,15 @@ class ConversationalAgent:
pairs_text += f"- {p.get('pair', 'N/A')}: ${p.get('liquidity', 'N/A'):,.0f} liquidity\n" pairs_text += f"- {p.get('pair', 'N/A')}: ${p.get('liquidity', 'N/A'):,.0f} liquidity\n"
response_text = f"**{symbol}** ({name})\n\nPrice: ${price}\nMarket Cap: ${mc:,.0f}\n24h Volume: ${vol:,.0f}{pairs_text}" response_text = f"**{symbol}** ({name})\n\nPrice: ${price}\nMarket Cap: ${mc:,.0f}\n24h Volume: ${vol:,.0f}{pairs_text}"
else: else:
response_text = f"Token not found: {address}" response_text = (
f"Token not found: {address}"
)
except json.JSONDecodeError:
response_text = "Failed to parse token data."
else: else:
response_text = f"Failed to get token details: {resp.status_code}" response_text = (
f"Failed to get token details: {output}"
)
return { return {
"response": response_text, "response": response_text,
@@ -417,39 +407,42 @@ class ConversationalAgent:
elif func_name == "get_price": elif func_name == "get_price":
token_ids = args.get("token_ids", "") token_ids = args.get("token_ids", "")
import os tokens_list = token_ids.replace(",", " ").split()
import asyncio if not tokens_list:
from ...core.config import get_settings response_text = "No token IDs provided."
else:
settings = get_settings() code, output = self._call_ave_script(
os.environ["AVE_API_KEY"] = settings.AVE_API_KEY "price", ["--tokens"] + tokens_list
os.environ["API_PLAN"] = settings.AVE_API_PLAN
from ave.http import api_post
tokens_list = [t.strip() for t in token_ids.split(",")]
resp = asyncio.run(
api_post(
"/tokens/price", {"token_ids": tokens_list}
) )
) if code == 0:
if resp.status_code == 200: try:
data = resp.json() data = json.loads(output)
prices = data.get("data", {}) prices = data.get("data", {})
if prices: if prices:
price_text = "**Token Prices:**\n" price_text = "**Token Prices:**\n"
for token_id, price_data in prices.items(): for (
price = price_data.get("price", "N/A") token_id,
price_data,
) in prices.items():
price = price_data.get(
"price", "N/A"
)
change_24h = price_data.get( change_24h = price_data.get(
"token_price_change_24h", "N/A" "token_price_change_24h", "N/A"
) )
price_text += f"- {token_id}: ${price} (24h: {change_24h}%)\n" price_text += f"- {token_id}: ${price} (24h: {change_24h}%)\n"
response_text = price_text response_text = price_text
else: else:
response_text = "No price data available." response_text = (
"No price data available."
)
except json.JSONDecodeError:
response_text = (
"Failed to parse price data."
)
else: else:
response_text = ( response_text = (
f"Failed to get prices: {resp.status_code}" f"Failed to get prices: {output}"
) )
return { return {
@@ -464,21 +457,12 @@ class ConversationalAgent:
address = args.get("address", "") address = args.get("address", "")
chain = args.get("chain", "bsc") chain = args.get("chain", "bsc")
import os code, output = self._call_ave_script(
import asyncio "risk", ["--address", address, "--chain", chain]
from ...core.config import get_settings
settings = get_settings()
os.environ["AVE_API_KEY"] = settings.AVE_API_KEY
os.environ["API_PLAN"] = settings.AVE_API_PLAN
from ave.http import api_get
resp = asyncio.run(
api_get(f"/contracts/{address}-{chain}")
) )
if resp.status_code == 200: if code == 0:
data = resp.json() try:
data = json.loads(output)
risk_data = data.get("data", {}) risk_data = data.get("data", {})
if risk_data: if risk_data:
is_honeypot = risk_data.get( is_honeypot = risk_data.get(
@@ -502,18 +486,16 @@ class ConversationalAgent:
): ):
risk_text += "\n⚠️ **Warning: High tax detected. Trade with caution!**" risk_text += "\n⚠️ **Warning: High tax detected. Trade with caution!**"
else: else:
risk_text += ( risk_text += "\n✅ This token appears safe to trade."
"\n✅ This token appears safe to trade."
)
response_text = risk_text response_text = risk_text
else: else:
response_text = ( response_text = (
f"No risk data available for {address}" f"No risk data available for {address}"
) )
except json.JSONDecodeError:
response_text = "Failed to parse risk data."
else: else:
response_text = ( response_text = f"Failed to get risk data: {output}"
f"Failed to get risk data: {resp.status_code}"
)
return { return {
"response": response_text, "response": response_text,
@@ -603,25 +585,21 @@ class ConversationalAgent:
keyword = args.get("keyword", "") keyword = args.get("keyword", "")
limit = args.get("limit", 10) limit = args.get("limit", 10)
# Execute the tool using ave-cloud-skill # Execute the tool using ave-cloud-skill CLI
import os code, output = self._call_ave_script(
import asyncio "search",
from ...core.config import get_settings [
"--keyword",
settings = get_settings() keyword,
os.environ["AVE_API_KEY"] = settings.AVE_API_KEY "--chain",
os.environ["API_PLAN"] = settings.AVE_API_PLAN "bsc",
"--limit",
from ave.http import api_get str(limit),
],
resp = asyncio.run(
api_get(
"/tokens",
{"keyword": keyword, "limit": limit, "chain": "bsc"},
) )
) if code == 0:
if resp.status_code == 200: try:
data = resp.json() data = json.loads(output)
tokens = data.get("data", {}).get("tokens", []) tokens = data.get("data", {}).get("tokens", [])
if tokens: if tokens:
token_list = "" token_list = ""
@@ -637,10 +615,10 @@ class ConversationalAgent:
response_text = f"Here are the search results for '{keyword}' on BSC:\n\n{token_list}\nWould you like me to set up a strategy for any of these?" response_text = f"Here are the search results for '{keyword}' on BSC:\n\n{token_list}\nWould you like me to set up a strategy for any of these?"
else: else:
response_text = f"No tokens found for '{keyword}'. Try a different keyword." response_text = f"No tokens found for '{keyword}'. Try a different keyword."
except json.JSONDecodeError:
response_text = "Failed to parse search results."
else: else:
response_text = ( response_text = f"Failed to search tokens: {output}"
f"Failed to search tokens: {resp.status_code}"
)
strategy_update = None strategy_update = None
@@ -666,29 +644,19 @@ class ConversationalAgent:
# Search for token if name is found # Search for token if name is found
if strategy_needs_confirmation and token_name: if strategy_needs_confirmation and token_name:
try: try:
from ..ave.client import AveCloudClient code, output = self._call_ave_script(
from ...core.config import get_settings "search",
["--keyword", token_name, "--chain", "bsc", "--limit", "5"],
settings = get_settings()
ave_client = AveCloudClient(
api_key=settings.AVE_API_KEY, plan=settings.AVE_API_PLAN
)
# Run async search in sync context
import asyncio
tokens = asyncio.run(
ave_client.get_tokens(
query=token_name, chain="bsc", limit=5
)
) )
if code == 0:
data = json.loads(output)
tokens = data.get("data", {}).get("tokens", [])
if tokens: if tokens:
token_search_results = [ token_search_results = [
{ {
"symbol": t.get("symbol", ""), "symbol": t.get("symbol", ""),
"name": t.get("name", ""), "name": t.get("name", ""),
"address": t.get( "address": t.get("token", ""),
"token", ""
), # trending API uses "token" for contract address
"chain": t.get("chain", "bsc"), "chain": t.get("chain", "bsc"),
} }
for t in tokens for t in tokens
@@ -1096,6 +1064,40 @@ Would you like me to adjust the strategy parameters based on these results?"""
print(f"Error updating strategy: {e}") print(f"Error updating strategy: {e}")
return False return False
def _call_ave_script(self, command: str, args: list) -> tuple[int, str]:
"""Call an ave-cloud-skill CLI script and return (status_code, stdout)."""
import json
import os
import subprocess
from ...core.config import get_settings
settings = get_settings()
repo_root = os.path.dirname(
os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
)
ave_skill_path = os.path.join(
repo_root, "ave-cloud-skill", "scripts", "ave_data_rest.py"
)
env = os.environ.copy()
env["AVE_API_KEY"] = settings.AVE_API_KEY
env["API_PLAN"] = settings.AVE_API_PLAN
env["AVE_USE_DOCKER"] = "false"
try:
result = subprocess.run(
["python3", ave_skill_path, command] + args,
capture_output=True,
text=True,
env=env,
timeout=30,
)
return result.returncode, result.stdout
except subprocess.TimeoutExpired:
return 1, "Error: Command timed out"
except Exception as e:
return 1, f"Error: {str(e)}"
def get_conversational_agent( def get_conversational_agent(
api_key: str = None, model: str = None, bot_id: str = None api_key: str = None, model: str = None, bot_id: str = None