Langchain CoinGecko Explorer Example

This example, Agentipy CoinGecko Explorer, is a simple, asynchronous tool that leverages Agentipy’s LangChain-based CoinGecko tools to fetch, process, and explore live cryptocurrency data. This tool demonstrates how to orchestrate multiple data-fetching tasks—including trending tokens, price data, detailed token info, and top gainers—using LangGraph, and how to integrate LLM analysis with OpenAI's GPT-3.5-turbo.

Features

  • Fetch Trending Tokens: Retrieves trending tokens data from CoinGecko.

  • Fetch Price Data: Extracts current price data for the top 5 tokens.

  • Token Information: Fetches detailed token info, with fallback logic if the token address is invalid.

  • Top Gainers: Retrieves the list of tokens with the highest gains in the last 24 hours.

  • LLM Analysis: Integrates OpenAI’s GPT-3.5-turbo to analyze token info and generate insights.

  • State Memory Logging: Maintains a log of workflow steps for robust debugging and traceability.

Code (coingecko_explorer.py)

import json
import asyncio
import openai
import re
from typing import TypedDict, List, Dict, Optional, Annotated
from langgraph.graph import StateGraph
from agentipy.agent import SolanaAgentKit
from agentipy.langchain.coingecko import get_coingecko_tools

# Set your OpenAI API key here or via environment variable
openai.api_key = ""

class AgentState(TypedDict):
    memory: Annotated[List[str], "multi"]
    trending_tokens: Optional[Dict]
    price_data: Optional[Dict]
    token_info: Optional[Dict]
    top_gainers: Optional[Dict]
    token_analysis: Optional[str]

solana_kit = SolanaAgentKit(
    private_key="",  # Use your private key
    rpc_url="https://api.mainnet-beta.solana.com"
)
tools = get_coingecko_tools(solana_kit)

def extract_trending_list(trending_tokens):
    """
    Helper to extract a list of tokens from trending_tokens.
    If trending_tokens is a dict with a 'coins' key, return that list;
    otherwise, assume trending_tokens is already a list.
    """
    if isinstance(trending_tokens, dict):
        return trending_tokens.get("coins", [])
    return trending_tokens

def is_valid_solana_address(addr: str) -> bool:
    """
    Simple heuristic: Solana addresses are Base58 strings typically 32-44 characters long.
    """
    if not (32 <= len(addr) <= 44):
        return False
    valid_chars = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"
    return all(c in valid_chars for c in addr)

async def fetch_trending_tokens(state: AgentState) -> AgentState:
    """Node: Fetch trending tokens"""
    tool = next(t for t in tools if t.name == "coingecko_get_trending_tokens")
    result = await tool._arun()

    state['memory'].append("Fetched trending tokens")
    state['trending_tokens'] = result['trending_tokens']
    return state

async def fetch_price_data(state: AgentState) -> AgentState:
    """Node: Fetch price data for trending tokens"""
    if not state['trending_tokens']:
        return state

    trending_list = extract_trending_list(state['trending_tokens'])
    if not trending_list:
        state['memory'].append("No tokens available for price data")
        return state

    tool = next(t for t in tools if t.name == "coingecko_get_token_price_data")
    addresses = [token["item"]["id"] for token in trending_list[:5]]
    input_json = json.dumps({"token_addresses": addresses})

    result = await tool._arun(input_json)
    state['memory'].append(f"Fetched price data for {len(addresses)} tokens")
    state['price_data'] = result['price_data']
    return state

async def fetch_token_info(state: AgentState) -> AgentState:
    """Node: Fetch info for first token"""
    if state['trending_tokens']:
        trending_list = extract_trending_list(state['trending_tokens'])
        if trending_list:
            token_item = trending_list[0]["item"]
            address = token_item.get("contract_address") or token_item.get("id")
            if not is_valid_solana_address(address):
                state['memory'].append(f"Token address '{address}' is not a valid Solana address. Using token item data instead.")
                state['token_info'] = token_item
                return state
            tool = next(t for t in tools if t.name == "coingecko_get_token_info")
            result = await tool._arun(json.dumps({"token_address": address}))
            state['memory'].append(f"Fetched token info for: {address}")
            state['token_info'] = result['token_info']
    return state

async def fetch_top_gainers(state: AgentState) -> AgentState:
    """Node: Fetch top gainers"""
    tool = next(t for t in tools if t.name == "coingecko_get_top_gainers")
    result = await tool._arun(json.dumps({"duration": "24h", "top_coins": 10}))

    state['memory'].append("Fetched top 24h gainers")
    state['top_gainers'] = result['top_gainers']
    return state

async def llm_interaction(state: AgentState) -> AgentState:
    """Node: Use OpenAI's LLM for interaction based on token info."""
    if not state['token_info']:
        state['memory'].append("No token info available to analyze with LLM.")
        return state

    prompt = f"Please analyze the following token info and provide insights:\n{json.dumps(state['token_info'], indent=2)}"

    try:
        response = await asyncio.to_thread(lambda: openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7
        ))
        analysis = response['choices'][0]['message']['content']
        state['memory'].append("LLM analysis completed.")
        state['token_analysis'] = analysis
    except Exception as e:
        state['memory'].append(f"LLM interaction failed: {str(e)}")

    return state

# Build graph with unique node keys
graph = StateGraph(AgentState)

graph.add_node("node_trending_tokens", fetch_trending_tokens)
graph.add_node("node_price_data", fetch_price_data)
graph.add_node("node_token_info", fetch_token_info)
graph.add_node("node_top_gainers", fetch_top_gainers)
graph.add_node("node_llm_interaction", llm_interaction)

graph.set_entry_point("node_trending_tokens")
graph.add_edge("node_trending_tokens", "node_price_data")
graph.add_edge("node_price_data", "node_token_info")
graph.add_edge("node_token_info", "node_top_gainers")
graph.add_edge("node_top_gainers", "node_llm_interaction")

app = graph.compile()

async def main():
    initial_state = AgentState(
        memory=[],
        trending_tokens=None,
        price_data=None,
        token_info=None,
        top_gainers=None,
        token_analysis=None
    )
    return await app.ainvoke(initial_state)

if __name__ == "__main__":
    final_state = asyncio.run(main())

    # Print trending tokens (showing the top 5)
    trending_list = extract_trending_list(final_state['trending_tokens'])
    print("Trending Tokens:")
    for token in trending_list[:5]:
        item = token.get("item", {})
        print(f"- ID: {item.get('id')}, Name: {item.get('name')}, Symbol: {item.get('symbol')}")

    print("\nPrice Data for 5 Tokens:")
    print(json.dumps(final_state['price_data'], indent=2))

    if final_state['token_info']:
        token_info = final_state['token_info']
        minimal_info = {
            "id": token_info.get("id"),
            "name": token_info.get("name"),
            "symbol": token_info.get("symbol"),
        }
        data = token_info.get("data", {})
        if data:
            minimal_info["price"] = data.get("price")
            minimal_info["market_cap"] = data.get("market_cap")
        print("\nMinimal Final Token Info:")
        print(json.dumps(minimal_info, indent=2))

    print("\nLLM Analysis:")
    print(final_state.get("token_analysis"))

    print("\nMemory History:")
    for entry in final_state['memory']:
        print(f"- {entry}")

Source Files

You can find the source files for this example (including its original readme.md with setup and usage instructions) on GitHub: https://github.com/niceberginc/agentipy/blob/main/examples/langChain/Agentipy_CoinGecko_Explorer/

PreviousLangchain CoinGecko Chatbot ExampleNextPyth Tool Examples