Security checks across malware telemetry and agentic risk
This is a disclosed live-crypto trading skill, but it requires persistent trading credentials and remote API trust for actions that can affect real funds.
Install only if you trust Katbot.ai with a Hyperliquid agent key. Prefer paper/testnet mode or a dedicated limited-permission agent wallet with minimal funds, keep auto-execution disabled, review any cron schedule before adding it, and remove the local identity directory if you stop using the skill.
return
if channel and target:
subprocess.run(
["openclaw", "message", "send",
"--channel", channel,
"--target", target,]
print(f"\n[Trigger] Starting workflow: {' '.join(workflow_cmd)}")
env = {**os.environ, "PYTHONPATH": _TOOLS_DIR}
subprocess.run(workflow_cmd, env=env)
elif args.dry_run:
print(
f"[DRY-RUN] Would start workflow with tokens: {tokens_for_workflow}"address = account.address
# Step 1: Get nonce
r = requests.get(f"{BASE_URL}/get-nonce/{address}?chain_id={CHAIN_ID}")
r.raise_for_status()
message_text = r.json()["message"]signature = signed.signature.hex()
# Step 3: Login
r = requests.post(f"{BASE_URL}/login", json={"address": address, "signature": signature, "chain_id": CHAIN_ID})
r.raise_for_status()
token_data = r.json()signature = signed.signature.hex()
# Step 3: Login
r = requests.post(f"{BASE_URL}/login", json={"address": address, "signature": signature, "chain_id": CHAIN_ID})
r.raise_for_status()
token_data = r.json()def list_portfolios(token: str) -> list:
"""List all portfolios for the authenticated user."""
r = requests.get(f"{BASE_URL}/portfolio", headers=_auth(token))
r.raise_for_status()
return r.json()if arbitrum_rpc_url is not None:
payload["arbitrum_rpc_url"] = arbitrum_rpc_url
r = requests.post(f"{BASE_URL}/portfolio", json=payload, headers=_auth(token))
r.raise_for_status()
return r.json()if window is not None:
params["window"] = window
r = requests.get(
f"{BASE_URL}/portfolio/{portfolio_id}",
params=params,
headers=_auth(token, agent_key)if max_history_messages is not None:
payload["max_history_messages"] = max_history_messages
r = requests.put(f"{BASE_URL}/portfolio/{portfolio_id}",
json=payload, headers=_auth(token))
r.raise_for_status()
return r.json()Returns:
List of symbol strings (e.g., ["BTC", "ETH", "SOL"])
"""
r = requests.get(f"{BASE_URL}/portfolio/{portfolio_id}/tokens", headers=_auth(token))
r.raise_for_status()
return r.json()Returns:
Dict with portfolio_id, chain_id, is_testnet, network_name
"""
r = requests.get(f"{BASE_URL}/portfolio/{portfolio_id}/chain-info", headers=_auth(token))
r.raise_for_status()
return r.json()Dict with timeseries list, portfolio_id, granularity, window, limit
"""
key = agent_private_key or AGENT_PRIVATE_KEY
r = requests.get(
f"{BASE_URL}/portfolio/{portfolio_id}/timeseries",
params={"granularity": granularity, "limit": limit, "window": window},
headers=_auth(token, key)Dict with status, result, and portfolio_id
"""
payload = {"action": action, "signature": signature, "nonce": nonce}
r = requests.post(f"{BASE_URL}/portfolio/{portfolio_id}/approve-builder-fee",
json=payload, headers=_auth(token))
r.raise_for_status()
return r.json()"""
key = agent_private_key or AGENT_PRIVATE_KEY
payload = {"agent_private_key": key, "is_testnet": is_testnet}
r = requests.post(f"{BASE_URL}/portfolio/validate-hyperliquid",
json=payload, headers=_auth(token))
r.raise_for_status()
return r.json()payload["agent_id"] = agent_id
if AGENT_PRIVATE_KEY:
payload["agent_private_key"] = AGENT_PRIVATE_KEY
r = requests.post(f"{BASE_URL}/agent/recommendation/message", json=payload, headers=_auth(token))
r.raise_for_status()
return r.json()if key:
payload["agent_private_key"] = key
r = requests.post(f"{BASE_URL}/agent/recommendation/response",
json=payload, headers=_auth(token))
r.raise_for_status()
return r.json()def get_recommendations(token: str, portfolio_id: int) -> list:
"""Get existing recommendations for a portfolio."""
r = requests.get(f"{BASE_URL}/portfolio/{portfolio_id}/recommendation", headers=_auth(token))
r.raise_for_status()
return r.json()payload["agent_private_key"] = AGENT_PRIVATE_KEY
if user_master_address:
payload["user_master_address"] = user_master_address
r = requests.post(f"{BASE_URL}/portfolio/{portfolio_id}/execute",
json=payload, headers=_auth(token))
r.raise_for_status()
return r.json()payload["user_master_address"] = user_master_address
if AGENT_PRIVATE_KEY:
payload["agent_private_key"] = AGENT_PRIVATE_KEY
r = requests.post(f"{BASE_URL}/portfolio/{portfolio_id}/close-position",
json=payload, headers=_auth(token))
r.raise_for_status()
return r.json()key = agent_private_key or AGENT_PRIVATE_KEY
if key:
params["agent_private_key"] = key
r = requests.get(f"{BASE_URL}/portfolio/{portfolio_id}/trade",
params=params, headers=_auth(token))
r.raise_for_status()
return r.json()if user_master_address:
params["user_master_address"] = user_master_address
key = agent_private_key or AGENT_PRIVATE_KEY
r = requests.get(f"{BASE_URL}/portfolio/{portfolio_id}/events",
params=params, headers=_auth(token, key))
r.raise_for_status()
return r.json()Returns:
List of AgentInfo dicts (id, name, max_history_messages, avatar_url, etc.)
"""
r = requests.get(f"{BASE_URL}/agents", headers=_auth(token))
r.raise_for_status()
return r.json()AgentInfo dict with id, name, avatar_url, etc.
"""
payload = {"name": name, "max_history_messages": max_history_messages}
r = requests.post(f"{BASE_URL}/agents", json=payload, headers=_auth(token))
r.raise_for_status()
return r.json()Returns:
AgentInfo dict
"""
r = requests.get(f"{BASE_URL}/agents/{agent_id}", headers=_auth(token))
r.raise_for_status()
return r.json()payload["max_history_messages"] = max_history_messages
if avatar_seed is not None:
payload["avatar_seed"] = avatar_seed
r = requests.put(f"{BASE_URL}/agents/{agent_id}", json=payload, headers=_auth(token))
r.raise_for_status()
return r.json()65/65 vendors flagged this skill as clean.