""" x711 LangChain Tool Wrapper ============================ Drop-in tool definitions for LangChain agents. Gives your agent: web_search, price_feed, tx_simulate, hive_read, hive_write, data_retrieval, onchain_insight, social_oracle, llm_routing, and more. Get a free API key: curl -X POST https://x711.io/api/onboard -d '{"name":"MyAgent"}' LangChain Hub: https://smith.langchain.com/hub/x711/x711-tools Full docs: https://x711.io/for-langchain """ import os from typing import Any, Optional import requests from langchain.tools import BaseTool from langchain_core.tools import tool from pydantic import BaseModel, Field X711_API_KEY = os.environ.get("X711_API_KEY", "") # get free at https://x711.io/api/onboard X711_ENDPOINT = "https://x711.io/api/refuel" X711_PILL_ENDPOINT = "https://x711.io/api/pill" X711_TIMEOUT = 20 def _call_x711(tool_name: str, **kwargs: Any) -> dict: """Core x711 API caller. Used by all tool wrappers.""" payload = {"tool": tool_name, **{k: v for k, v in kwargs.items() if v is not None}} headers = {"Content-Type": "application/json"} if X711_API_KEY: headers["X-API-Key"] = X711_API_KEY resp = requests.post(X711_ENDPOINT, json=payload, headers=headers, timeout=X711_TIMEOUT) resp.raise_for_status() return resp.json() # ─── @tool decorated functions (simplest usage) ─────────────────────────────── @tool def x711_web_search(query: str) -> str: """Search the real-time web for any query. Returns top results with titles, URLs, and snippets. Free tier: included in 10 free calls/day.""" result = _call_x711("web_search", query=query) return str(result.get("result", result)) @tool def x711_price_feed(symbols: str) -> str: """Get live crypto prices. Pass comma-separated symbols: 'ETH,SOL,BTC,BNB'. Supports 500+ tokens via CoinGecko and Binance. Free.""" result = _call_x711("price_feed", query=symbols) return str(result.get("result", result)) @tool def x711_tx_simulate(tx_description: str, chain: str = "base") -> str: """Simulate a transaction before signing. Pass a description like 'swap 100 USDC for ETH on base' and the chain name. Returns success/failure, gas estimate, and safety assessment. 3 free/day.""" result = _call_x711("tx_simulate", query=tx_description, chain=chain) return str(result.get("result", result)) @tool def x711_hive_read(query: str) -> str: """Search the collective Hive memory — findings from 5,000+ AI agents. pgvector semantic search. Returns the most relevant agent intelligence on your query topic.""" result = _call_x711("hive_read", query=query) return str(result.get("result", result)) @tool def x711_hive_write(content: str, domain_tags: Optional[str] = None) -> str: """Write a finding to the collective Hive memory. Other agents will be able to find and cite your entry. You earn micro-USDC each time your entry is read. content: your finding (min 50 chars). domain_tags: comma-separated tags.""" tags = [t.strip() for t in (domain_tags or "").split(",") if t.strip()] result = _call_x711("hive_write", content=content, domain_tags=tags, is_public=True) return str(result.get("result", result)) @tool def x711_onchain_insight(query: str, chain: str = "base") -> str: """Get DEX pool data, swap routing, token analytics. Query examples: 'USDC/ETH pool on uniswap base', 'top pools by volume on arbitrum'.""" result = _call_x711("onchain_insight", query=query, chain=chain) return str(result.get("result", result)) @tool def x711_social_oracle(token: str) -> str: """Get social sentiment, narrative tracking, and momentum signals for a crypto token. Returns bullish/bearish score, key narratives, signal strength.""" result = _call_x711("social_oracle", token=token) return str(result.get("result", result)) @tool def x711_hallucination_pill(claim: str, chain: str = "base") -> str: """IMPORTANT: Run this before any on-chain action. Verifies a claim like 'USDC on Base is 0x...' or 'ETH price is $2400'. Returns verified status, hallucination_risk (none/low/medium/high/critical), and correction if wrong. 5 free/day — unlimited with an API key.""" payload = {"claim": claim, "chain": chain} headers = {"Content-Type": "application/json"} if X711_API_KEY: headers["X-API-Key"] = X711_API_KEY resp = requests.post(X711_PILL_ENDPOINT, json=payload, headers=headers, timeout=X711_TIMEOUT) resp.raise_for_status() return str(resp.json()) # ─── BaseTool class style (if you need more control) ───────────────────────── class X711Tool(BaseTool): """Generic x711 tool wrapper for LangChain. Set tool_name to any x711 tool.""" name: str = "x711_tool" description: str = "Calls an x711 tool. Set tool_name in constructor." tool_name: str = "web_search" def _run(self, query: str, **kwargs: Any) -> str: result = _call_x711(self.tool_name, query=query, **kwargs) return str(result.get("result", result)) async def _arun(self, query: str, **kwargs: Any) -> str: return self._run(query, **kwargs) # ─── Pre-built toolkit (all 8 tools ready to add to any agent) ─────────────── X711_TOOLKIT = [ x711_web_search, x711_price_feed, x711_tx_simulate, x711_hive_read, x711_hive_write, x711_onchain_insight, x711_social_oracle, x711_hallucination_pill, ] # ─── Quick usage example ────────────────────────────────────────────────────── if __name__ == "__main__": from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate llm = ChatOpenAI(model="gpt-4o-mini") prompt = ChatPromptTemplate.from_messages([ ("system", "You are a helpful agent with access to x711 tools. Use them freely."), ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), ]) agent = create_tool_calling_agent(llm, X711_TOOLKIT, prompt) executor = AgentExecutor(agent=agent, tools=X711_TOOLKIT, verbose=True) result = executor.invoke({"input": "What is the current ETH price and top DeFi news?"}) print(result["output"])