Elfaai Examples
These examples demonstrate the use of the Agentipy library to interact with the Elfa AI API for blockchain and crypto-related data analysis. Agentipy simplifies interacting with various APIs, including the Elfa AI API, by providing a structured approach for authentication, tool creation, and asynchronous execution.
The examples show how to:
Initialize a
SolanaAgentKit
object.Create and use Elfa AI tools provided by Agentipy.
Call different Elfa AI endpoints to retrieve information such as smart mentions, top mentions by ticker, search mentions by keywords, trending tokens, and Twitter account stats.
One script (
elfaai2.py
) provides an interactive way to choose and run these analyses.
Code
elfaai1.py
(Basic Elfa AI API Calls)
elfaai1.py
(Basic Elfa AI API Calls)import asyncio
import json
import os
from agentipy.agent import SolanaAgentKit
from agentipy.langchain.elfaai import get_elfaai_tools
async def main():
private_key = os.getenv("SOLANA_PRIVATE_KEY")
elfa_ai_api_key = os.getenv("ELFA_AI_API_KEY")
if not private_key or not elfa_ai_api_key:
print("ERROR: Missing required environment variables!")
print("Run the following in your terminal:")
print('export SOLANA_PRIVATE_KEY="your_base58_private_key_here"')
print('export ELFA_AI_API_KEY="your_elfa_ai_api_key_here"')
return
try:
solana_kit = SolanaAgentKit(private_key=private_key, elfa_ai_api_key=elfa_ai_api_key)
except Exception as e:
print("Error initializing SolanaAgentKit:", e)
return
tools = get_elfaai_tools(solana_kit)
results = {}
# Get Smart Mentions
smart_mentions_tool = next(tool for tool in tools if tool.name == "elfa_ai_get_smart_mentions")
smart_mentions_input = json.dumps({"limit": 5, "offset": 0})
results["Smart Mentions"] = await smart_mentions_tool._arun(smart_mentions_input)
# Get Top Mentions by Ticker
top_mentions_tool = next(tool for tool in tools if tool.name == "elfa_ai_get_top_mentions_by_ticker")
top_mentions_input = json.dumps({
"ticker": "SOL",
"time_window": "1h",
"page": 1,
"page_size": 5,
"include_account_details": True
})
results["Top Mentions by Ticker"] = await top_mentions_tool._arun(top_mentions_input)
# Search Mentions by Keywords
search_mentions_tool = next(tool for tool in tools if tool.name == "elfa_ai_search_mentions_by_keywords")
search_mentions_input = json.dumps({
"keywords": "Solana blockchain",
"from_timestamp": 1672531200,
"to_timestamp": 1740787200,
"limit": 5
})
results["Search Mentions by Keywords"] = await search_mentions_tool._arun(search_mentions_input)
# Get Trending Tokens
trending_tokens_tool = next(tool for tool in tools if tool.name == "elfa_ai_get_trending_tokens")
trending_tokens_input = json.dumps({
"time_window": "24h",
"page": 1,
"page_size": 10,
"min_mentions": 5
})
results["Trending Tokens"] = await trending_tokens_tool._arun(trending_tokens_input)
# Get Smart Twitter Account Stats
twitter_stats_tool = next(tool for tool in tools if tool.name == "elfa_ai_get_smart_twitter_account_stats")
twitter_stats_input = json.dumps({
"username": "solana"
})
results["Twitter Stats"] = await twitter_stats_tool._arun(twitter_stats_input)
# Print results
for name, result in results.items():
print(f"\n{name}:\n", json.dumps(result, indent=2))
if __name__ == "__main__":
asyncio.run(main())
elfaai2.py
(Interactive Elfa AI Analysis)
elfaai2.py
(Interactive Elfa AI Analysis)import asyncio
import json
import os
import datetime
from agentipy.agent import SolanaAgentKit
from agentipy.langchain.elfaai import get_elfaai_tools
async def main():
# Load required environment variables
private_key = os.getenv("SOLANA_PRIVATE_KEY")
elfa_ai_api_key = os.getenv("ELFA_AI_API_KEY")
if not private_key or not elfa_ai_api_key:
print("ERROR: Missing required environment variables!")
print('Run the following in your terminal:')
print('export SOLANA_PRIVATE_KEY="your_base58_private_key_here"')
print('export ELFA_AI_API_KEY="your_elfa_ai_api_key_here"')
return
try:
solana_kit = SolanaAgentKit(private_key=private_key, elfa_ai_api_key=elfa_ai_api_key)
except Exception as e:
print("Error initializing SolanaAgentKit:", e)
return
# Load tools from Elfa AI
tools = get_elfaai_tools(solana_kit)
# Let user select which analysis to run
print("Choose an analysis option:")
print("1. Get Smart Mentions")
print("2. Get Top Mentions by Ticker")
print("3. Search Mentions by Keywords (with date range)")
print("4. Get Trending Tokens")
print("5. Get Twitter Stats")
choice = input("Enter your choice (1-5): ").strip()
results = {}
if choice == "1":
# Get Smart Mentions
limit = input("Enter number of smart mentions to retrieve (default 5): ").strip() or "5"
offset = input("Enter offset (default 0): ").strip() or "0"
smart_mentions_input = json.dumps({
"limit": int(limit),
"offset": int(offset)
})
tool = next(tool for tool in tools if tool.name == "elfa_ai_get_smart_mentions")
results["Smart Mentions"] = await tool._arun(smart_mentions_input)
elif choice == "2":
# Get Top Mentions by Ticker
ticker = input("Enter token ticker (e.g., SOL): ").strip() or "SOL"
time_window = input("Enter time window (e.g., 1h): ").strip() or "1h"
page = input("Enter page number (default 1): ").strip() or "1"
page_size = input("Enter page size (default 5): ").strip() or "5"
include_details_input = input("Include account details? (yes/no, default no): ").strip().lower()
# Convert input to Boolean (True if user says yes, else False)
include_details = True if include_details_input.startswith("y") else False
top_mentions_input = json.dumps({
"ticker": ticker,
"time_window": time_window,
"page": int(page),
"page_size": int(page_size),
"include_account_details": include_details
})
tool = next(tool for tool in tools if tool.name == "elfa_ai_get_top_mentions_by_ticker")
results["Top Mentions by Ticker"] = await tool._arun(top_mentions_input)
elif choice == "3":
# Search Mentions by Keywords with a date range
keywords = input("Enter keywords for search: ").strip()
from_date_str = input("Enter start date (YYYY-MM-DD): ").strip()
to_date_str = input("Enter end date (YYYY-MM-DD): ").strip()
try:
from_timestamp = int(datetime.datetime.strptime(from_date_str, "%Y-%m-%d").timestamp())
to_timestamp = int(datetime.datetime.strptime(to_date_str, "%Y-%m-%d").timestamp())
except Exception as e:
print("Error parsing dates:", e)
return
# Validate date range: must be at least 1 day and at most 30 days
one_day = 86400
thirty_days = 30 * one_day
diff = to_timestamp - from_timestamp
if diff < one_day or diff > thirty_days:
print("Error: The date range must be between 1 and 30 days.")
return
# Get limit input and enforce allowed range (20-30)
limit_input = input("Enter number of results to retrieve (default 5): ").strip()
try:
requested_limit = int(limit_input) if limit_input else 5
except ValueError:
requested_limit = 5
if requested_limit < 20:
print("Note: Limit must be between 20 and 30. Using minimum value of 20.")
requested_limit = 20
elif requested_limit > 30:
print("Note: Limit must be between 20 and 30. Using maximum value of 30.")
requested_limit = 30
search_mentions_input = json.dumps({
"keywords": keywords,
"from_timestamp": from_timestamp,
"to_timestamp": to_timestamp,
"limit": requested_limit,
"cursor": ""
})
tool = next(tool for tool in tools if tool.name == "elfa_ai_search_mentions_by_keywords")
results["Search Mentions by Keywords"] = await tool._arun(search_mentions_input)
elif choice == "4":
# Get Trending Tokens
time_window = input("Enter time window (default 24h): ").strip() or "24h"
page = input("Enter page number (default 1): ").strip() or "1"
page_size = input("Enter number of tokens to show (default 10): ").strip() or "10"
min_mentions = input("Enter minimum mentions (default 5): ").strip() or "5"
trending_tokens_input = json.dumps({
"time_window": time_window,
"page": int(page),
"page_size": int(page_size),
"min_mentions": int(min_mentions)
})
tool = next(tool for tool in tools if tool.name == "elfa_ai_get_trending_tokens")
results["Trending Tokens"] = await tool._arun(trending_tokens_input)
elif choice == "5":
# Get Twitter Stats
username = input("Enter Twitter username: ").strip()
twitter_stats_input = json.dumps({
"username": username
})
tool = next(tool for tool in tools if tool.name == "elfa_ai_get_smart_twitter_account_stats")
results["Twitter Stats"] = await tool._arun(twitter_stats_input)
else:
print("Invalid choice!")
return
for name, result in results.items():
print(f"\n{name}:\n{json.dumps(result, indent=2)}")
if __name__ == "__main__":
asyncio.run(main())
Source Files
You can find the source files for this example (including readme.md
with setup and usage instructions) on GitHub: https://github.com/niceberginc/agentipy/blob/main/examples/Elfaai/