Python Integration

Access Hive Intelligence using Python's requests library.

Installation

pip install requests

Requirements:

• Python 3.8+
• requests library

Quick Start

import requests

API_URL = "https://api.hiveintelligence.xyz/api/execute"

def call_hive(tool_name: str, arguments: dict) -> dict:
    """Execute a Hive Intelligence tool."""
    response = requests.post(API_URL, json={
        "toolName": tool_name,
        "arguments": arguments
    })
    response.raise_for_status()
    return response.json()

# Get Bitcoin price
result = call_hive("get_price", {
    "ids": "bitcoin",
    "vs_currencies": "usd"
})
print(result)
# {'data': {'bitcoin': {'usd': 67234.00}}}

Helper Class

Create a reusable client class:

import requests
from typing import Any

class HiveClient:
    """Simple client for Hive Intelligence API."""

    def __init__(self, base_url: str = "https://api.hiveintelligence.xyz"):
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({"Content-Type": "application/json"})

    def execute(self, tool_name: str, arguments: dict = None) -> dict:
        """Execute a tool and return the response."""
        response = self.session.post(
            f"{self.base_url}/api/execute",
            json={
                "toolName": tool_name,
                "arguments": arguments or {}
            }
        )
        response.raise_for_status()
        return response.json()

    def get_price(self, coin_id: str, currency: str = "usd") -> float:
        """Get price for a single coin."""
        result = self.execute("get_price", {
            "ids": coin_id,
            "vs_currencies": currency
        })
        return result["data"][coin_id][currency]

    def check_security(self, contract: str, chain_id: str = "1") -> dict:
        """Check token security."""
        result = self.execute("get_token_security", {
            "contract_addresses": contract,
            "chain_id": chain_id
        })
        return result["data"]

# Usage
client = HiveClient()
btc_price = client.get_price("bitcoin")
print(f"Bitcoin: ${btc_price}")

Working with Tool Categories

Market Data

client = HiveClient()

# Get top coins by market cap
top_coins = client.execute("get_coins_market_data", {
    "vs_currency": "usd",
    "order": "market_cap_desc",
    "per_page": 100
})

# Get historical price data
history = client.execute("get_coin_market_chart_range", {
    "id": "bitcoin",
    "vs_currency": "usd",
    "from": 1704067200,  # Unix timestamp
    "to": 1706745600
})

# Get trending coins
trending = client.execute("get_trending", {})

DeFi Analytics

# Get all protocols
protocols = client.execute("get_protocols", {})

# Get specific protocol TVL
protocol = client.execute("get_defi_protocol", {
    "protocol": "aave"
})

# Get yield pools
yields = client.execute("get_yield_pools", {
    "chain": "ethereum"
})

Portfolio Tracking

# Get wallet balances
balances = client.execute("get_wallet_balance", {
    "address": "0x..."
})

# Get DeFi positions
positions = client.execute("get_wallet_defi_positions", {
    "address": "0x..."
})

# Get transaction history
history = client.execute("get_wallet_history", {
    "address": "0x...",
    "limit": 100
})

Security Analysis

# Check token security
security = client.execute("get_token_security", {
    "contract_addresses": "0x...",
    "chain_id": "1"
})

if security["data"].get("is_honeypot"):
    print("WARNING: Potential honeypot!")

if float(security["data"].get("sell_tax", "0")) > 10:
    print(f"WARNING: High sell tax: {security['data']['sell_tax']}%")

Async Support

For concurrent requests, use aiohttp:

import asyncio
import aiohttp

API_URL = "https://api.hiveintelligence.xyz/api/execute"

async def call_hive_async(session, tool_name: str, arguments: dict) -> dict:
    async with session.post(API_URL, json={
        "toolName": tool_name,
        "arguments": arguments
    }) as response:
        return await response.json()

async def main():
    async with aiohttp.ClientSession() as session:
        # Concurrent requests
        btc_task = call_hive_async(session, "get_price", {
            "ids": "bitcoin", "vs_currencies": "usd"
        })
        eth_task = call_hive_async(session, "get_price", {
            "ids": "ethereum", "vs_currencies": "usd"
        })

        btc_price, eth_price = await asyncio.gather(btc_task, eth_task)

        print(f"BTC: ${btc_price['data']['bitcoin']['usd']}")
        print(f"ETH: ${eth_price['data']['ethereum']['usd']}")

asyncio.run(main())

Error Handling

import requests

def call_hive_safe(tool_name: str, arguments: dict) -> dict:
    try:
        response = requests.post(
            "https://api.hiveintelligence.xyz/api/execute",
            json={"toolName": tool_name, "arguments": arguments},
            timeout=30
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 429:
            print("Rate limited. Wait and retry.")
        elif e.response.status_code == 400:
            print(f"Bad request: {e.response.text}")
        else:
            print(f"HTTP error: {e}")
        raise
    except requests.exceptions.Timeout:
        print("Request timed out")
        raise
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}")
        raise

Complete Example

import requests

API_URL = "https://api.hiveintelligence.xyz/api/execute"

def call_hive(tool_name: str, arguments: dict) -> dict:
    response = requests.post(API_URL, json={
        "toolName": tool_name,
        "arguments": arguments
    })
    response.raise_for_status()
    return response.json()

def analyze_token(contract_address: str, chain_id: str = "1"):
    """Analyze a token's security and basic info."""

    # Check security
    security = call_hive("get_token_security", {
        "contract_addresses": contract_address,
        "chain_id": chain_id
    })["data"]

    print(f"Contract: {contract_address}")
    print(f"Chain ID: {chain_id}")
    print("-" * 40)
    print(f"Honeypot: {security.get('is_honeypot', 'Unknown')}")
    print(f"Open Source: {security.get('is_open_source', 'Unknown')}")
    print(f"Buy Tax: {security.get('buy_tax', '0')}%")
    print(f"Sell Tax: {security.get('sell_tax', '0')}%")
    print(f"Holder Count: {security.get('holder_count', 'Unknown')}")

    # Risk assessment
    risks = []
    if security.get("is_honeypot"):
        risks.append("CRITICAL: Honeypot detected")
    if security.get("is_mintable"):
        risks.append("HIGH: Token is mintable")
    if security.get("hidden_owner"):
        risks.append("HIGH: Hidden owner")
    if float(security.get("sell_tax", "0")) > 10:
        risks.append(f"MEDIUM: High sell tax ({security['sell_tax']}%)")

    if risks:
        print("\nRisks Found:")
        for risk in risks:
            print(f"  - {risk}")
    else:
        print("\nNo major risks detected")

if __name__ == "__main__":
    # Analyze PEPE token
    analyze_token("0x6982508145454Ce325dDbE47a25d4ec3d2311933")

Quick Reference

Full Tools Reference →