Elfa AI Python SDK

Official Python SDK for the Elfa API v2 — social intelligence, AI chat, and the Auto condition engine for crypto. Sync and async clients, fully typed with Pydantic.

Features

• Social intelligence — trending tokens, mentions, narratives, smart stats, event summaries
• AI chat — market analysis and conversational chat via client.chat, streamed via client.chat_stream
• Auto condition engine — build EQL queries that notify or trade via client.auto
• Sync and async — ElfaClient and AsyncElfaClient, same surface
• Typed — Pydantic v2 models, full type hints
• Robust — retries with backoff, typed errors, HMAC request signing

The SDK returns processed metadata and tweet links only — never raw tweet content. For raw tweets, call the X (Twitter) API directly using the returned links/ids.

Installation

pip install elfa-sdk

Quick start

Synchronous

from elfa import ElfaClient

client = ElfaClient(api_key="your-api-key")

trending = client.get_trending_tokens(time_window="24h")
for token in trending.data.data:
    print(token.token, token.current_count, f"{token.change_percent:+.1f}%")

mentions = client.get_keyword_mentions(keywords="bitcoin,ethereum", time_window="1h")
for mention in mentions.data:
    print(mention.link, mention.like_count)

answer = client.chat("What's the sentiment on Bitcoin today?")
print(answer.data.message)

Asynchronous

import asyncio
from elfa import AsyncElfaClient

async def main():
    async with AsyncElfaClient(api_key="your-api-key") as client:
        stats = await client.get_account_smart_stats("elonmusk")
        print(stats.data.smart_following_count)

asyncio.run(main())

Configuration

client = ElfaClient(
    api_key="your-api-key",
    base_url="https://api.elfa.ai",  # default (production)
    timeout=30.0,                    # per-request timeout, seconds
    retries=3,                       # retries for idempotent (GET) requests
    retry_delay=1.0,                 # base delay for exponential backoff
    hmac_secret=None,                # required for Auto trade-action queries (see below)
    headers=None,                    # extra headers sent on every request
)

# Quick reachability/auth check
assert client.test_connection() is True

The Auto engine is also constructable standalone if that is all you need:

from elfa import AutoClient

auto = AutoClient(api_key="your-api-key", hmac_secret="your-hmac-secret")
auto.close()

The API key is sent as the x-elfa-api-key header on every request. Read it from the environment in your app:

import os
from elfa import ElfaClient

client = ElfaClient(api_key=os.environ["ELFA_API_KEY"])

Core data & chat

All methods exist on both ElfaClient (sync) and AsyncElfaClient (async).

Time-ranged endpoints accept either time_window="24h" or both from_time and to_time (unix seconds).

Streaming chat (SSE)

chat_stream takes the same arguments as chat and yields one event per data: frame, ending on the terminating [DONE] frame. Requires a PAYG or Enterprise API key.

for event in client.chat_stream("What is the sentiment on SOL?"):
    if event.type == "text":
        print(event.content, end="")
    elif event.type == "complete":
        print("\ncredits:", event.creditsConsumed)

# async
async for event in async_client.chat_stream("What is the sentiment on SOL?"):
    print(event.type)

Event types are session_info, title, text, text_complete, status, credits, complete, invalid_request and error. Payload fields vary by type and are preserved as model extras.

Auto condition engine (client.auto)

Build EQL queries that watch conditions and fire actions (notify, webhook, or trade). Notification-only queries need no secret; trade-action queries require an hmac_secret.

query = {
    "query": {
        "conditions": {
            "AND": [{
                "source": "price", "method": "current",
                "args": {"symbol": "BTC", "exchange": "hyperliquid"},
                "operator": ">", "value": 250000,
            }]
        },
        "actions": [{"stepId": "notify", "type": "notify", "params": {"message": "BTC > 250k"}}],
        "expiresIn": "24h",
    },
    "title": "btc breakout alert",
}

client.auto.validate_query(query)
created = client.auto.create_query(query)
query_id = created.id or created.query_id

status = client.auto.get_query(query_id)
client.auto.cancel_query(query_id)
client.auto.delete_query(query_id)

Also available: chat, list_queries, drafts (list_drafts/get_draft/upsert_draft/delete_draft/validate_draft/convert_draft), list_sessions/get_session, list_executions/get_execution, exchanges (list_exchanges/connect_exchange/disconnect_exchange), and validate_symbol.

Streaming notifications (SSE)

for event in client.auto.stream_query(query_id):
    print(event.event, event.data)

# async
async for event in async_client.auto.stream_all():
    print(event.event, event.data)

HMAC signing

Auto trade-action queries are signed when hmac_secret is set. The SDK builds the signature over timestamp + METHOD + mounted_path + body and sends x-elfa-timestamp and x-elfa-signature headers. Signing every mutation is safe, so passing hmac_secret is always fine. Generate a secret in the dev portal.

Error handling

from elfa import (
    ElfaAPIError,
    ElfaAuthenticationError,
    ElfaRateLimitError,
    ElfaValidationError,
    ElfaNetworkError,
)

try:
    client.get_trending_tokens(time_window="24h")
except ElfaAuthenticationError:
    ...  # bad/missing API key
except ElfaRateLimitError as e:
    print("retry after", e.retry_after, "reset", e.reset_time)
except ElfaValidationError as e:
    print("invalid params", e.validation_errors)
except ElfaNetworkError:
    ...  # connection problem
except ElfaAPIError as e:
    print("api error", e.status_code, e)

Idempotent (GET) requests are retried with exponential backoff on network errors, rate limits, and 5xx responses. Mutations are not retried automatically.

Development

git clone https://github.com/elfa-ai/elfa-sdk-python.git
cd elfa-sdk-python
pip install -e ".[dev]"

make check   # flake8 + mypy + pytest
make format  # black + isort

Live integration tests run only when ELFA_API_KEY is set (optionally ELFA_BASE_URL, ELFA_HMAC_SECRET); otherwise they skip.

Support

• Documentation
• Issues

License

MIT — see LICENSE.