Python Integration

pip install requests

import os
import requests

BASE_URL = "https://mcp.hiveintelligence.xyz"
API_KEY = os.environ["HIVE_API_KEY"]

def execute(tool: str, args: dict | None = None) -> dict:
    response = requests.post(
        f"{BASE_URL}/api/v1/execute",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"tool": tool, "args": args or {}},
        timeout=30,
    )
    response.raise_for_status()
    return response.json()

price = execute("get_price", {
    "ids": "bitcoin",
    "vs_currencies": "usd",
})

print(price["bitcoin"]["usd"])

from __future__ import annotations

import requests


class HiveClient:
    def __init__(self, api_key: str, base_url: str = "https://mcp.hiveintelligence.xyz") -> None:
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({"Authorization": f"Bearer {api_key}"})

    def execute(self, tool: str, args: dict | None = None) -> dict:
        response = self.session.post(
            f"{self.base_url}/api/v1/execute",
            json={"tool": tool, "args": args or {}},
            timeout=30,
        )
        response.raise_for_status()
        return response.json()

    def list_tools(self, limit: int = 50) -> dict:
        response = self.session.get(
            f"{self.base_url}/api/v1/tools",
            params={"limit": limit},
            timeout=30,
        )
        response.raise_for_status()
        return response.json()


client = HiveClient(api_key="YOUR_HIVE_API_KEY")

market = client.execute("get_coins_market_data", {
    "vs_currency": "usd",
    "order": "market_cap_desc",
    "per_page": 5,
})

wallet = client.execute("moralis_get_wallet_net_worth", {
    "address": "0x1234...",
    "chain": "eth",
})

security = client.execute("get_token_security", {
    "contract_addresses": "0x6982508145454Ce325dDbE47a25d4ec3d2311933",
    "chainId": "1",
})

pools = client.execute("get_yield_pools", {
    "chain": "ethereum",
})

positions = client.execute("moralis_get_wallet_defi_positions", {
    "address": "0x1234...",
    "chain": "eth",
})

markets = client.execute("codex_prediction_markets", {
    "networkId": 1,
})

import time
import requests


class HiveError(Exception):
    def __init__(self, code: str, message: str, http_status: int) -> None:
        super().__init__(f"{code}: {message}")
        self.code = code
        self.message = message
        self.http_status = http_status


def execute_with_retry(client: HiveClient, tool: str, args: dict, max_retries: int = 3) -> dict:
    for attempt in range(max_retries):
        try:
            response = client.session.post(
                f"{client.base_url}/api/v1/execute",
                json={"tool": tool, "args": args or {}},
                timeout=30,
            )
            if response.status_code == 429:
                # Respect Retry-After header (seconds); fall back to exponential.
                sleep_for = int(response.headers.get("Retry-After", 2 ** attempt))
                time.sleep(sleep_for)
                continue
            if response.status_code >= 500:
                time.sleep(2 ** attempt)
                continue
            response.raise_for_status()
            return response.json()
        except requests.RequestException as exc:
            if attempt == max_retries - 1:
                raise HiveError("network_error", str(exc), 0) from exc
            time.sleep(2 ** attempt)
    raise HiveError("exhausted_retries", "All retries failed", 0)

import asyncio
import httpx


class AsyncHiveClient:
    def __init__(self, api_key: str, base_url: str = "https://mcp.hiveintelligence.xyz") -> None:
        self.base_url = base_url
        self._client = httpx.AsyncClient(
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=httpx.Timeout(30.0, connect=5.0),
            limits=httpx.Limits(max_connections=32),
        )

    async def execute(self, tool: str, args: dict | None = None) -> dict:
        response = await self._client.post(
            f"{self.base_url}/api/v1/execute",
            json={"tool": tool, "args": args or {}},
        )
        response.raise_for_status()
        return response.json()

    async def aclose(self) -> None:
        await self._client.aclose()


async def daily_brief(api_key: str) -> dict:
    client = AsyncHiveClient(api_key)
    try:
        prices, tvl, oi = await asyncio.gather(
            client.execute("get_price", {"ids": "bitcoin,ethereum", "vs_currencies": "usd"}),
            client.execute("get_protocol_tvl", {}),
            client.execute("get_open_interest", {"exchange": "binance"}),
        )
        return {"prices": prices, "tvl": tvl[:5], "oi": oi}
    finally:
        await client.aclose()

def discover_tools(client: HiveClient, search: str | None = None) -> list[dict]:
    cursor, items = None, []
    while True:
        params = {"limit": 200}
        if cursor:
            params["cursor"] = cursor
        if search:
            params["search"] = search
        page = client.session.get(
            f"{client.base_url}/api/v1/tools",
            params=params,
            timeout=30,
        ).json()
        items.extend(page.get("data", []))
        cursor = page.get("meta", {}).get("cursor")
        if not cursor:
            return items

wallet_tools = discover_tools(client, search="wallet")

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic

client = MultiServerMCPClient({
    "hive": {
        "transport": "streamable_http",
        "url": "https://mcp.hiveintelligence.xyz/mcp",
        "headers": {"Authorization": "Bearer $HIVE_API_KEY"},
    }
})
tools = await client.get_tools()
agent = create_react_agent(ChatAnthropic(model="claude-sonnet-4-5"), tools)