LangChain

This guide shows how to integrate Epanya with LangChain by creating custom BaseTool subclasses. Once defined, these tools drop into any AgentExecutor or LCEL chain — they're indistinguishable from built-in LangChain tools.

Installation

pip install epanya langchain langchain-openai langchain-community

Custom tools

Subclass BaseTool for each Epanya operation. LangChain uses the description attribute to decide when to call each tool, so write it clearly for the LLM.

import asyncio
import json
from typing import Any, Optional, Type
from langchain.tools import BaseTool
from pydantic import BaseModel, Field
from epanya import EpanyaClient, create_test_signer


# ── Shared client ──────────────────────────────────────────────────────────────

AGENT_WALLET = "0xf39Fd6e51aad88F6F4ce6aB8827279cffFb92266"  # replace with real wallet

client = EpanyaClient(
    wallet_address=AGENT_WALLET,
    api_url="https://api.epanya.ai",
    signer=create_test_signer(AGENT_WALLET),  # swap for real signer in production
)


# ── discover_products ─────────────────────────────────────────────────────────

class DiscoverInput(BaseModel):
    q:         Optional[str]   = Field(None, description="Free-text search query")
    category:  Optional[str]   = Field(None, description="One of: compute, datasets, apis, models, tools, physical, agent_labor")
    max_price: Optional[float] = Field(None, description="Maximum price in USDC")
    sort:      Optional[str]   = Field(None, description="One of: price_asc, price_desc, rating, newest")
    limit:     Optional[int]   = Field(10,   description="Max results to return")


class EpanyaDiscoverTool(BaseTool):
    name:        str = "discover_products"
    description: str = (
        "Search the Epanya marketplace for products. Use this to find compute resources, "
        "APIs, datasets, AI models, tools, or agent labor services. "
        "Returns a list of matching products with names, USDC prices, and IDs. "
        "You must call this before purchase_product."
    )
    args_schema: Type[BaseModel] = DiscoverInput

    def _run(self, q=None, category=None, max_price=None, sort=None, limit=10) -> str:
        return asyncio.get_event_loop().run_until_complete(
            self._arun(q=q, category=category, max_price=max_price, sort=sort, limit=limit)
        )

    async def _arun(self, q=None, category=None, max_price=None, sort=None, limit=10) -> str:
        products = await client.discover(
            q=q,
            category=category,
            max_price=max_price,
            sort=sort,
            limit=limit or 10,
        )
        if not products:
            return "No products found matching those criteria."
        lines = []
        for i, p in enumerate(products, 1):
            lines.append(
                f"{i}. {p['name']} — ${p['priceUsdc']} USDC ({p['pricingModel']})\n"
                f"   ID: {p['id']}\n"
                f"   {p['description']}"
            )
        return "\n\n".join(lines)


# ── purchase_product ──────────────────────────────────────────────────────────

class PurchaseInput(BaseModel):
    product_id: str = Field(..., description="The product UUID from discover_products results")


class EpanyaPurchaseTool(BaseTool):
    name:        str = "purchase_product"
    description: str = (
        "Purchase a product from the Epanya marketplace by its ID. "
        "Only call this after using discover_products and selecting the best option. "
        "The payment is made automatically in USDC via the x402 protocol. "
        "Returns the transaction ID and the service endpoint URL."
    )
    args_schema: Type[BaseModel] = PurchaseInput

    def _run(self, product_id: str) -> str:
        return asyncio.get_event_loop().run_until_complete(self._arun(product_id))

    async def _arun(self, product_id: str) -> str:
        result = await client.purchase(product_id)
        return json.dumps({
            "transaction_id": result["transactionId"],
            "endpoint_url":   result["product"]["endpointUrl"],
            "amount_paid":    str(result["amount"]) + " USDC",
            "status":         "escrowed",
        })


# ── check_budget ──────────────────────────────────────────────────────────────

class EpanyaBudgetTool(BaseTool):
    name:        str = "check_budget"
    description: str = (
        "Check how much USDC this agent has spent and whether the budget limit "
        "has been reached. Call this before purchasing if unsure about remaining budget."
    )

    def _run(self) -> str:
        return asyncio.get_event_loop().run_until_complete(self._arun())

    async def _arun(self) -> str:
        budget = await client.check_budget()
        return json.dumps({
            "spent":           str(budget["spent"]) + " USDC",
            "budget_limit":    str(budget.get("budgetLimit") or "unlimited") + " USDC",
            "remaining":       str(budget.get("remaining")  or "unlimited") + " USDC",
            "budget_exceeded": budget["budgetExceeded"],
        })

Agent setup

Pass the tools to a LangChain agent. The example uses create_openai_tools_agent with GPT-4o, but any tool-calling model works.

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Tools
tools = [
    EpanyaDiscoverTool(),
    EpanyaPurchaseTool(),
    EpanyaBudgetTool(),
]

# Prompt
prompt = ChatPromptTemplate.from_messages([
    (
        "system",
        "You are an autonomous procurement agent with access to the Epanya marketplace. "
        "When asked to procure something:\n"
        "1. Check the budget if there is any uncertainty about remaining funds.\n"
        "2. Search for products matching the requirements.\n"
        "3. Evaluate the options — weigh price, ratings, and fit.\n"
        "4. Purchase the best option and return the service endpoint URL.",
    ),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

# Agent + executor
agent          = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

Run the agent:

result = agent_executor.invoke({
    "input": (
        "I need a GPU inference endpoint for image generation. "
        "Budget: $3 per request maximum. Find the best option and purchase it."
    )
})

print(result["output"])

With verbose=True, LangChain prints each tool call and result as the agent works through the task:

> Entering new AgentExecutor chain...
Invoking: `check_budget` with `{}`
{"spent": "4.20 USDC", "budget_limit": "100.00 USDC", "remaining": "95.80 USDC", ...}
Invoking: `discover_products` with `{"category": "compute", "max_price": 3.0, "sort": "rating"}`
1. NovaCorp H100 Inference — $2.50 USDC (per_request)
   ID: 7a3f9c12-...
   ...
Invoking: `purchase_product` with `{"product_id": "7a3f9c12-..."}`
{"transaction_id": "tx_...", "endpoint_url": "https://...", "amount_paid": "2.50 USDC"}
Purchased NovaCorp H100 Inference for $2.50 USDC. Endpoint: https://...

LCEL chain variant

If you're using LangChain Expression Language (LCEL) instead of AgentExecutor, bind the tools to the model and use agent.stream():

from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

llm_with_tools = llm.bind_tools(tools)

# Simple LCEL chain (no loop — single turn only)
chain = (
    {"input": RunnablePassthrough()}
    | prompt
    | llm_with_tools
)

# For a full agentic loop with LCEL, use create_openai_tools_agent above.
# The chain approach works well for single-step tool use.
response = chain.invoke({"input": "What compute products are under $1?", "agent_scratchpad": []})
print(response)

Full code

import asyncio, json
from typing import Optional, Type
from langchain.tools import BaseTool
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from pydantic import BaseModel, Field
from epanya import EpanyaClient, create_test_signer

WALLET = "0xf39Fd6e51aad88F6F4ce6aB8827279cffFb92266"
client = EpanyaClient(wallet_address=WALLET, api_url="https://api.epanya.ai", signer=create_test_signer(WALLET))

def run(coro): return asyncio.get_event_loop().run_until_complete(coro)

class DiscoverInput(BaseModel):
    q: Optional[str] = None; category: Optional[str] = None
    max_price: Optional[float] = None; limit: Optional[int] = 10

class EpanyaDiscoverTool(BaseTool):
    name = "discover_products"
    description = "Search Epanya marketplace. Returns product names, prices, and IDs."
    args_schema: Type[BaseModel] = DiscoverInput
    def _run(self, q=None, category=None, max_price=None, limit=10):
        ps = run(client.discover(q=q, category=category, max_price=max_price, limit=limit or 10))
        return "\n".join(f"{i+1}. {p['name']} ${p['priceUsdc']} — {p['id']}" for i, p in enumerate(ps)) or "No results."

class PurchaseInput(BaseModel):
    product_id: str = Field(..., description="Product UUID")

class EpanyaPurchaseTool(BaseTool):
    name = "purchase_product"
    description = "Purchase a product by ID. Returns endpoint URL."
    args_schema: Type[BaseModel] = PurchaseInput
    def _run(self, product_id: str):
        r = run(client.purchase(product_id))
        return json.dumps({"endpoint": r["product"]["endpointUrl"], "tx": r["transactionId"], "paid": str(r["amount"]) + " USDC"})

class EpanyaBudgetTool(BaseTool):
    name = "check_budget"; description = "Check USDC spend vs budget."
    def _run(self):
        b = run(client.check_budget())
        return json.dumps({"spent": b["spent"], "limit": b.get("budgetLimit","unlimited"), "exceeded": b["budgetExceeded"]})

llm    = ChatOpenAI(model="gpt-4o", temperature=0)
tools  = [EpanyaDiscoverTool(), EpanyaPurchaseTool(), EpanyaBudgetTool()]
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a procurement agent. Use your tools to find and purchase marketplace products."),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])
executor = AgentExecutor(agent=create_openai_tools_agent(llm, tools, prompt), tools=tools, verbose=True)

if __name__ == "__main__":
    out = executor.invoke({"input": "Find the cheapest translation API and buy it."})
    print(out["output"])