Deploy Financial AI Pipes with Langbase

{
  "name": "purple-flea-trader",
  "description": "Autonomous crypto trading agent on Purple Flea",
  "model": "openai:gpt-4o",
  "temperature": 0.2,
  "max_tokens": 2048,
  "systemPrompt": "You are a financial AI agent on Purple Flea (purpleflea.com). You manage a crypto portfolio across 275 markets, can play provably fair casino games, and send payments via trustless escrow. Always check balances before trading. Max 5% portfolio risk per trade. Log all actions with timestamps and reasoning.",
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "get_portfolio",
        "description": "Get current portfolio balances and positions",
        "parameters": {"type": "object", "properties": {}}
      }
    },
    {
      "type": "function",
      "function": {
        "name": "get_market_price",
        "description": "Get price and 24h stats for a trading pair",
        "parameters": {
          "type": "object",
          "properties": {
            "symbol": {"type": "string", "description": "Trading pair e.g. BTC-USD"}
          },
          "required": ["symbol"]
        }
      }
    },
    {
      "type": "function",
      "function": {
        "name": "place_trade",
        "description": "Execute a market order on Purple Flea",
        "parameters": {
          "type": "object",
          "properties": {
            "symbol": {"type": "string"},
            "side": {"type": "string", "enum": ["buy", "sell"]},
            "usdc_amount": {"type": "number"}
          },
          "required": ["symbol", "side", "usdc_amount"]
        }
      }
    },
    {
      "type": "function",
      "function": {
        "name": "escrow_payment",
        "description": "Send USDC via Purple Flea trustless escrow",
        "parameters": {
          "type": "object",
          "properties": {
            "recipient_id": {"type": "string"},
            "usdc_amount": {"type": "number"},
            "memo": {"type": "string"}
          },
          "required": ["recipient_id", "usdc_amount"]
        }
      }
    }
  ],
  "memorysets": ["purple-flea-transaction-history"],
  "secrets": [
    {"name": "PURPLE_FLEA_KEY", "value": "your-api-key-here"}
  ]
}

from langbase import Langbase
import json
from datetime import datetime

lb = Langbase(api_key="your-langbase-key")

# Create the transaction history Memoryset
memoryset = lb.memory.create(
    name="purple-flea-transaction-history",
    description="Trade history, market signals, and portfolio snapshots for the Purple Flea trading agent",
    embedding_model="openai:text-embedding-3-small"
)

def log_trade(symbol, side, amount, price, reasoning):
    """Log a completed trade to Memoryset for future context."""
    doc = {
        "timestamp": datetime.utcnow().isoformat(),
        "action": "trade",
        "symbol": symbol,
        "side": side,
        "amount_usd": amount,
        "execution_price": price,
        "reasoning": reasoning
    }
    text = (
        f"Trade: {side.upper()} ${amount} of {symbol} at ${price}. "
        f"Reason: {reasoning}. Time: {doc['timestamp']}"
    )
    lb.memory.documents.create(
        memory_name="purple-flea-transaction-history",
        content=text,
        metadata=doc
    )

def log_market_signal(symbol, signal_type, value, context):
    """Store a market signal for pattern recognition."""
    text = (
        f"Signal: {signal_type} on {symbol} = {value}. "
        f"Context: {context}. Time: {datetime.utcnow().isoformat()}"
    )
    lb.memory.documents.create(
        memory_name="purple-flea-transaction-history",
        content=text,
        metadata={"type": "signal", "symbol": symbol}
    )

# Seed with initial portfolio snapshot
log_market_signal("PORTFOLIO", "initial-snapshot", "agent started",
    "Agent initialized with Purple Flea faucet USDC, ready to trade")

import os, json
import requests
from langbase import Langbase

lb = Langbase(api_key=os.environ["LANGBASE_API_KEY"])
PF_KEY = os.environ["PURPLE_FLEA_KEY"]
PF = "https://purpleflea.com/api"
pf = requests.Session()
pf.headers["Authorization"] = f"Bearer {PF_KEY}"

def execute_tool(name: str, args: dict):
    """Route tool calls to Purple Flea API endpoints."""
    if name == "get_portfolio":
        return pf.get(f"{PF}/portfolio").json()
    elif name == "get_market_price":
        return pf.get(f"{PF}/market/{args['symbol']}/ticker").json()
    elif name == "place_trade":
        result = pf.post(f"{PF}/trading/market", json=args).json()
        # Log to Memoryset for future context
        log_trade(
            args["symbol"], args["side"], args["usdc_amount"],
            result.get("execution_price"),
            args.get("reason", "automated rebalance")
        )
        return result
    elif name == "escrow_payment":
        return pf.post("https://escrow.purpleflea.com/pay", json=args).json()

def run_pipe(task: str):
    """Invoke the Langbase Pipe and process tool calls."""
    messages = [{"role": "user", "content": task}]
    thread_id = None

    while True:
        response = lb.pipe.run(
            name="purple-flea-trader",
            messages=messages,
            thread_id=thread_id
        )
        thread_id = response.get("threadId")
        completion = response["completion"]

        if not completion.get("tool_calls"):
            return completion["content"]

        # Execute each tool call, collect results
        tool_results = []
        for call in completion["tool_calls"]:
            fn = call["function"]
            args = json.loads(fn["arguments"])
            result = execute_tool(fn["name"], args)
            tool_results.append({
                "tool_call_id": call["id"],
                "role": "tool",
                "content": json.dumps(result)
            })

        messages = tool_results  # feed results back

# Run the agent on a daily task
result = run_pipe(
    "Review my portfolio, check if any position needs rebalancing, "
    "and if I have idle USDC above $200 put $100 into BTC. "
    "Use my transaction history from memory to avoid repeating trades."
)
print(result)