langchain_tool.py

"""LangChain tool that fetches trading signals via x402 payments."""

import asyncio
from typing import Optional, Type

from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field

from x402_agent_kit import SignalFuse


class SignalInput(BaseModel):
    symbol: str = Field(description="Cryptocurrency symbol, e.g. BTC, ETH, SOL")


class SignalFuseTool(BaseTool):
    name: str = "signalfuse_signal"
    description: str = (
        "Fetch the latest trading signal for a cryptocurrency symbol. "
        "Pays for the API call with USDC on Base via x402."
    )
    args_schema: Type[BaseModel] = SignalInput

    wallet_key: str

    def _run(self, symbol: str) -> dict:
        sf = SignalFuse(wallet_key=self.wallet_key)
        return asyncio.run(sf.get_signal(symbol))

    async def _arun(self, symbol: str) -> dict:
        sf = SignalFuse(wallet_key=self.wallet_key)
        return await sf.get_signal(symbol)


# Usage with a LangChain agent:
#
#   from langchain.agents import initialize_agent, AgentType
#   from langchain_openai import ChatOpenAI
#
#   tool = SignalFuseTool(wallet_key="0xYOUR_KEY")
#   llm = ChatOpenAI()
#   agent = initialize_agent([tool], llm, agent=AgentType.OPENAI_FUNCTIONS)
#   agent.run("What is the signal for ETH?")