Natural Language Trade CLI

This guide builds an interactive REPL where you type plain English commands like "swap 0.5 ETH to USDC on base" or "show my portfolio" and get formatted responses. It demonstrates the A2A (Agent-to-Agent) protocol's natural language communication.

What the CLI Does

1. Loads your API key from environment variables

2. Starts an interactive REPL with a suwappu> prompt

3. Sends your natural language input to Suwappu via A2A message/send

4. Parses the task response — if completed, pretty-prints artifacts

5. If working or submitted, polls with tasks/get and shows a spinner

6. Supports Ctrl+C to cancel running tasks via tasks/cancel

7. Keeps a local task history accessible with the history command

Python Version

class=class="hl-str">"hl-comment">#!/usr/bin/env python3
class="hl-str">""class="hl-str">"
Suwappu Natural Language Trade CLI — Python
Interactive REPL for communicating with Suwappu via the A2A protocol.
"class="hl-str">""

import os
import sys
import json
import time
import signal
import requests

A2A_URL = class="hl-str">"https:class="hl-commentclass="hl-str">">//api.suwappu.bot/a2a"

class=class="hl-str">"hl-comment"># Spinner frames for polling animation
SPINNER = [class="hl-str">"⠋", class="hl-str">"⠙", class="hl-str">"⠹", class="hl-str">"⠸", class="hl-str">"⠼", class="hl-str">"⠴", class="hl-str">"⠦", class="hl-str">"⠧", class="hl-str">"⠇", class="hl-str">"⠏"]

class=class="hl-str">"hl-comment"># Track state
request_id = 0
task_history = []
current_task_id = None


def next_id():
    class="hl-str">""class="hl-str">"Generate incrementing JSON-RPC request IDs."class="hl-str">""
    global request_id
    request_id += 1
    return request_id


def a2a_request(headers, method, params):
    class="hl-str">""class="hl-str">"Send a JSON-RPC 2.0 request to the A2A endpoint."class="hl-str">""
    payload = {
        class="hl-str">"jsonrpc": class="hl-str">"2.0",
        class="hl-str">"id": next_id(),
        class="hl-str">"method": method,
        class="hl-str">"params": params,
    }
    response = requests.post(A2A_URL, headers=headers, json=payload)
    response.raise_for_status()
    data = response.json()

    if class="hl-str">"error" in data:
        raise Exception(fclass="hl-str">"A2A error {data[class="hl-str">'error'][class="hl-str">'code']}: {data[class="hl-str">'error'][class="hl-str">'message']}")

    return data[class="hl-str">"result"]


def send_message(headers, text):
    class="hl-str">""class="hl-str">"Send a natural language message via message/send."class="hl-str">""
    return a2a_request(headers, class="hl-str">"message/send", {
        class="hl-str">"message": {
            class="hl-str">"role": class="hl-str">"user",
            class="hl-str">"parts": [{class="hl-str">"type": class="hl-str">"text", class="hl-str">"text": text}],
        }
    })


def get_task(headers, task_id):
    class="hl-str">""class="hl-str">"Poll a task by ID via tasks/get."class="hl-str">""
    return a2a_request(headers, class="hl-str">"tasks/get", {class="hl-str">"taskId": task_id})


def cancel_task(headers, task_id):
    class="hl-str">""class="hl-str">"Cancel a running task via tasks/cancel."class="hl-str">""
    return a2a_request(headers, class="hl-str">"tasks/cancel", {class="hl-str">"taskId": task_id})


def format_artifacts(artifacts):
    class="hl-str">""class="hl-str">"Pretty-print task artifacts."class="hl-str">""
    output = []
    for artifact in artifacts:
        for part in artifact.get(class="hl-str">"parts", []):
            if part[class="hl-str">"type"] == class="hl-str">"text":
                output.append(part[class="hl-str">"text"])
            elif part[class="hl-str">"type"] == class="hl-str">"data":
                output.append(json.dumps(part[class="hl-str">"data"], indent=2))
    return class="hl-str">"\n".join(output)


def poll_task(headers, task_id):
    class="hl-str">""class="hl-str">"Poll a task until it reaches a terminal state, showing a spinner."class="hl-str">""
    global current_task_id
    current_task_id = task_id
    frame = 0

    try:
        while True:
            result = get_task(headers, task_id)
            task = result[class="hl-str">"task"]
            state = task[class="hl-str">"status"][class="hl-str">"state"]

            if state == class="hl-str">"completed":
                class=class="hl-str">"hl-comment"># Clear spinner line
                sys.stdout.write(class="hl-str">"\r" + class="hl-str">" " * 40 + class="hl-str">"\r")
                if task.get(class="hl-str">"artifacts"):
                    print(format_artifacts(task[class="hl-str">"artifacts"]))
                else:
                    print(task[class="hl-str">"status"].get(class="hl-str">"message", class="hl-str">"Done."))
                return task

            elif state in (class="hl-str">"failed", class="hl-str">"canceled"):
                sys.stdout.write(class="hl-str">"\r" + class="hl-str">" " * 40 + class="hl-str">"\r")
                message = task[class="hl-str">"status"].get(class="hl-str">"message", state.capitalize())
                print(fclass="hl-str">"Task {state}: {message}")
                return task

            class=class="hl-str">"hl-comment"># Show spinner
            sys.stdout.write(fclass="hl-str">"\r  {SPINNER[frame % len(SPINNER)]} Processing...")
            sys.stdout.flush()
            frame += 1
            time.sleep(1)

    finally:
        current_task_id = None


def handle_response(headers, result):
    class="hl-str">""class="hl-str">"Handle a message/send response — print immediately or poll."class="hl-str">""
    task = result[class="hl-str">"task"]
    state = task[class="hl-str">"status"][class="hl-str">"state"]
    task_id = task[class="hl-str">"id"]

    class=class="hl-str">"hl-comment"># Save to history
    task_history.append({
        class="hl-str">"id": task_id,
        class="hl-str">"state": state,
        class="hl-str">"timestamp": task[class="hl-str">"status"].get(class="hl-str">"timestamp", class="hl-str">""),
    })

    if state == class="hl-str">"completed":
        if task.get(class="hl-str">"artifacts"):
            print(format_artifacts(task[class="hl-str">"artifacts"]))
        else:
            print(task[class="hl-str">"status"].get(class="hl-str">"message", class="hl-str">"Done."))
    elif state in (class="hl-str">"submitted", class="hl-str">"working"):
        poll_task(headers, task_id)
    elif state == class="hl-str">"failed":
        message = task[class="hl-str">"status"].get(class="hl-str">"message", class="hl-str">"Unknown error")
        print(fclass="hl-str">"Failed: {message}")
    else:
        print(fclass="hl-str">"Unexpected state: {state}")


def print_history():
    class="hl-str">""class="hl-str">"Print local task history."class="hl-str">""
    if not task_history:
        print(class="hl-str">"No task history yet.")
        return

    print(fclass="hl-str">"\n  {class="hl-str">'class="hl-commentclass="hl-str">">#':<4} {class="hl-str">'Task ID':<40} {class="hl-str">'State':<12} {class="hl-str">'Time'}")
    print(fclass="hl-str">"  {class="hl-str">'-' * 70}")
    for i, entry in enumerate(task_history, 1):
        print(fclass="hl-str">"  {i:<4} {entry[class="hl-str">'id']:<40} {entry[class="hl-str">'state']:<12} {entry[class="hl-str">'timestamp']}")
    print()


def print_help():
    class="hl-str">""class="hl-str">"Print help text."class="hl-str">""
    print(class="hl-str">""class="hl-str">"
  Suwappu Natural Language CLI
  ────────────────────────────
  Type any natural language command. Examples:

    swap 0.5 ETH to USDC on base
    price of ETH
    prices for ETH, BTC, SOL
    show my portfolio on ethereum
    list supported chains
    quote 100 USDC to WBTC

  Special commands:
    help      Show this help message
    history   Show task history
    quit      Exit the CLI

  Press Ctrl+C during a running task to cancel it.
"class="hl-str">"")


def setup_signal_handler(headers):
    class="hl-str">""class="hl-str">"Set up Ctrl+C handler to cancel running tasks."class="hl-str">""
    def handler(sig, frame):
        global current_task_id
        if current_task_id:
            sys.stdout.write(class="hl-str">"\r" + class="hl-str">" " * 40 + class="hl-str">"\r")
            print(class="hl-str">"Canceling task...")
            try:
                cancel_task(headers, current_task_id)
                print(class="hl-str">"Task canceled.")
            except Exception as e:
                print(fclass="hl-str">"Cancel failed: {e}")
            current_task_id = None
        else:
            print(class="hl-str">"\nGoodbye!")
            sys.exit(0)

    signal.signal(signal.SIGINT, handler)


def main():
    api_key = os.environ.get(class="hl-str">"SUWAPPU_API_KEY")
    if not api_key:
        print(class="hl-str">"Error: Set SUWAPPU_API_KEY environment variable.")
        print(class="hl-str">"  export SUWAPPU_API_KEY=suwappu_sk_your_api_key")
        sys.exit(1)

    headers = {
        class="hl-str">"Authorization": fclass="hl-str">"Bearer {api_key}",
        class="hl-str">"Content-Type": class="hl-str">"application/json",
    }

    setup_signal_handler(headers)
    print_help()

    while True:
        try:
            user_input = input(class="hl-str">"suwappu> ").strip()
        except EOFError:
            print(class="hl-str">"\nGoodbye!")
            break

        if not user_input:
            continue

        class=class="hl-str">"hl-comment"># Handle special commands
        lower = user_input.lower()
        if lower == class="hl-str">"quit" or lower == class="hl-str">"exit":
            print(class="hl-str">"Goodbye!")
            break
        elif lower == class="hl-str">"help":
            print_help()
            continue
        elif lower == class="hl-str">"history":
            print_history()
            continue

        class=class="hl-str">"hl-comment"># Send to A2A
        try:
            result = send_message(headers, user_input)
            handle_response(headers, result)
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429:
                print(class="hl-str">"Rate limited. Wait a moment and try again.")
            else:
                print(fclass="hl-str">"HTTP error: {e}")
        except Exception as e:
            print(fclass="hl-str">"Error: {e}")

        print()  class=class="hl-str">"hl-comment"># Blank line between responses


if __name__ == class="hl-str">"__main__":
    main()

Running the Python Version

-str">"hl-comment"># Install dependencies
-kw">pip install requests

-str">"hl-comment"># Set your API key
-kw">export SUWAPPU_API_KEY=suwappu_sk_your_api_key

-str">"hl-comment"># Start the CLI
python natural_language_cli.py

Example session:

suwappu> price of ETH
ETH: $3,500.42 (+2.5% 24h)

suwappu> swap 0.5 ETH to USDC on base
Quote ready: 0.5 ETH -> 1,247.50 USDC on Base
{
  class="hl-str">"quote_id": class="hl-str">"q_abc123",
  class="hl-str">"from_token": class="hl-str">"ETH",
  class="hl-str">"to_token": class="hl-str">"USDC",
  class="hl-str">"from_amount": class="hl-str">"0.5",
  class="hl-str">"to_amount": class="hl-str">"1247.50",
  class="hl-str">"chain": class="hl-str">"base"
}

suwappu> list supported chains
Suwappu supports: ethereum, base, arbitrum, optimism, polygon, bsc, solana

suwappu> history

  class=class="hl-str">"hl-comment">#    Task ID                                  State        Time
  ----------------------------------------------------------------------
  1    a1b2c3d4-e5f6-7890-abcd-ef1234567890     completed    2026-03-08T12:00:00Z
  2    b2c3d4e5-f6a7-8901-bcde-f12345678901     completed    2026-03-08T12:00:05Z
  3    c3d4e5f6-a7b8-9012-cdef-123456789012     completed    2026-03-08T12:00:10Z

suwappu> quit
Goodbye!

---

TypeScript Version

class=class="hl-str">"hl-comment">#!/usr/bin/env npx tsx
/**
 * Suwappu Natural Language Trade CLI — TypeScript
 * Interactive REPL for communicating with Suwappu via the A2A protocol.
 */

import * as readline from class="hl-str">"readline";

const A2A_URL = class="hl-str">"https:class="hl-commentclass="hl-str">">//api.suwappu.bot/a2a";
const SPINNER = [class="hl-str">"⠋", class="hl-str">"⠙", class="hl-str">"⠹", class="hl-str">"⠸", class="hl-str">"⠼", class="hl-str">"⠴", class="hl-str">"⠦", class="hl-str">"⠧", class="hl-str">"⠇", class="hl-str">"⠏"];

let requestId = 0;
const taskHistory: Array<{ id: string; state: string; timestamp: string }> = [];
let currentTaskId: string | null = null;

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

function nextId(): number {
  return ++requestId;
}

async function a2aRequest(apiKey: string, method: string, params: Record<string, unknown>) {
  const response = await fetch(A2A_URL, {
    method: class="hl-str">"POST",
    headers: {
      class="hl-str">"Content-Type": class="hl-str">"application/json",
      Authorization: Bearer ${apiKey},
    },
    body: JSON.stringify({
      jsonrpc: class="hl-str">"2.0",
      id: nextId(),
      method,
      params,
    }),
  });

  if (!response.ok) {
    throw new Error(HTTP ${response.status}: ${await response.text()});
  }

  const data = await response.json();
  if (data.error) {
    throw new Error(A2A error ${data.error.code}: ${data.error.message});
  }
  return data.result;
}

async function sendMessage(apiKey: string, text: string) {
  return a2aRequest(apiKey, class="hl-str">"message/send", {
    message: {
      role: class="hl-str">"user",
      parts: [{ type: class="hl-str">"text", text }],
    },
  });
}

async function getTask(apiKey: string, taskId: string) {
  return a2aRequest(apiKey, class="hl-str">"tasks/get", { taskId });
}

async function cancelTask(apiKey: string, taskId: string) {
  return a2aRequest(apiKey, class="hl-str">"tasks/cancel", { taskId });
}

function formatArtifacts(artifacts: Array<{ parts: Array<{ type: string; text?: string; data?: unknown }> }>): string {
  const output: string[] = [];
  for (const artifact of artifacts) {
    for (const part of artifact.parts ?? []) {
      if (part.type === class="hl-str">"text" && part.text) {
        output.push(part.text);
      } else if (part.type === class="hl-str">"data" && part.data) {
        output.push(JSON.stringify(part.data, null, 2));
      }
    }
  }
  return output.join(class="hl-str">"\n");
}

async function pollTask(apiKey: string, taskId: string) {
  currentTaskId = taskId;
  let frame = 0;

  try {
    while (true) {
      const result = await getTask(apiKey, taskId);
      const task = result.task;
      const state = task.status.state;

      if (state === class="hl-str">"completed") {
        process.stdout.write(class="hl-str">"\r" + class="hl-str">" ".repeat(40) + class="hl-str">"\r");
        if (task.artifacts?.length) {
          console.log(formatArtifacts(task.artifacts));
        } else {
          console.log(task.status.message ?? class="hl-str">"Done.");
        }
        return task;
      }

      if (state === class="hl-str">"failed" || state === class="hl-str">"canceled") {
        process.stdout.write(class="hl-str">"\r" + class="hl-str">" ".repeat(40) + class="hl-str">"\r");
        console.log(Task ${state}: ${task.status.message ?? state});
        return task;
      }

      process.stdout.write(\r  ${SPINNER[frame % SPINNER.length]} Processing...);
      frame++;
      await sleep(1000);
    }
  } finally {
    currentTaskId = null;
  }
}

function handleResponse(apiKey: string, result: { task: any }) {
  const task = result.task;
  const state = task.status.state;

  taskHistory.push({
    id: task.id,
    state,
    timestamp: task.status.timestamp ?? class="hl-str">"",
  });

  if (state === class="hl-str">"completed") {
    if (task.artifacts?.length) {
      console.log(formatArtifacts(task.artifacts));
    } else {
      console.log(task.status.message ?? class="hl-str">"Done.");
    }
    return Promise.resolve();
  }

  if (state === class="hl-str">"submitted" || state === class="hl-str">"working") {
    return pollTask(apiKey, task.id);
  }

  if (state === class="hl-str">"failed") {
    console.log(Failed: ${task.status.message ?? class="hl-str">"Unknown error"});
  } else {
    console.log(Unexpected state: ${state});
  }

  return Promise.resolve();
}

function printHistory() {
  if (taskHistory.length === 0) {
    console.log(class="hl-str">"No task history yet.");
    return;
  }

  console.log(\n  ${class="hl-str">"class="hl-commentclass="hl-str">">#".padEnd(4)} ${class="hl-str">"Task ID".padEnd(40)} ${class="hl-str">"State".padEnd(12)} Time);
  console.log(  ${class="hl-str">"-".repeat(70)});
  taskHistory.forEach((entry, i) => {
    console.log(
        ${String(i + 1).padEnd(4)} ${entry.id.padEnd(40)} ${entry.state.padEnd(12)} ${entry.timestamp}
    );
  });
  console.log();
}

function printHelp() {
  console.log(

  Suwappu Natural Language CLI
  ────────────────────────────
  Type any natural language command. Examples:

    swap 0.5 ETH to USDC on base
    price of ETH
    prices for ETH, BTC, SOL
    show my portfolio on ethereum
    list supported chains
    quote 100 USDC to WBTC

  Special commands:
    help      Show this help message
    history   Show task history
    quit      Exit the CLI

  Press Ctrl+C during a running task to cancel it.
);
}

async function main() {
  const apiKey = process.env.SUWAPPU_API_KEY;
  if (!apiKey) {
    console.error(class="hl-str">"Error: Set SUWAPPU_API_KEY environment variable.");
    process.exit(1);
  }

  class=class="hl-str">"hl-comment">// Handle Ctrl+C for task cancellation
  process.on(class="hl-str">"SIGINT", async () => {
    if (currentTaskId) {
      process.stdout.write(class="hl-str">"\r" + class="hl-str">" ".repeat(40) + class="hl-str">"\r");
      console.log(class="hl-str">"Canceling task...");
      try {
        await cancelTask(apiKey, currentTaskId);
        console.log(class="hl-str">"Task canceled.");
      } catch (e) {
        console.error(Cancel failed: ${e});
      }
      currentTaskId = null;
    } else {
      console.log(class="hl-str">"\nGoodbye!");
      process.exit(0);
    }
  });

  const rl = readline.createInterface({
    input: process.stdin,
    output: process.stdout,
    prompt: class="hl-str">"suwappu> ",
  });

  printHelp();
  rl.prompt();

  rl.on(class="hl-str">"line", async (line) => {
    const input = line.trim();
    if (!input) {
      rl.prompt();
      return;
    }

    const lower = input.toLowerCase();
    if (lower === class="hl-str">"quit" || lower === class="hl-str">"exit") {
      console.log(class="hl-str">"Goodbye!");
      rl.close();
      process.exit(0);
    }
    if (lower === class="hl-str">"help") {
      printHelp();
      rl.prompt();
      return;
    }
    if (lower === class="hl-str">"history") {
      printHistory();
      rl.prompt();
      return;
    }

    try {
      const result = await sendMessage(apiKey, input);
      await handleResponse(apiKey, result);
    } catch (error: unknown) {
      const message = error instanceof Error ? error.message : String(error);
      if (message.includes(class="hl-str">"429")) {
        console.log(class="hl-str">"Rate limited. Wait a moment and try again.");
      } else {
        console.error(Error: ${message});
      }
    }

    console.log();
    rl.prompt();
  });

  rl.on(class="hl-str">"close", () => {
    console.log(class="hl-str">"\nGoodbye!");
    process.exit(0);
  });
}

main().catch(console.error);

Running the TypeScript Version

-str">"hl-comment"># Install tsx for running TypeScript directly
-kw">npm install -g tsx

-str">"hl-comment"># Set your API key
-kw">export SUWAPPU_API_KEY=suwappu_sk_your_api_key

-str">"hl-comment"># Start the CLI
npx tsx natural_language_cli.ts

---

Customization Tips

Add Wallet Context

To automatically include your wallet address in portfolio queries, prepend it to every message:

WALLET = os.environ.get(class="hl-str">"WALLET_ADDRESS", class="hl-str">"")

def send_with_context(headers, text):
    if WALLET and class="hl-str">"portfolio" in text.lower():
        text = fclass="hl-str">"{text} for {WALLET}"
    return send_message(headers, text)

Persistent History

Save task history to a JSON file so it persists across sessions:

import json
from pathlib import Path

HISTORY_FILE = Path.home() / class="hl-str">".suwappu_history.json"

def load_history():
    if HISTORY_FILE.exists():
        return json.loads(HISTORY_FILE.read_text())
    return []

def save_history(history):
    HISTORY_FILE.write_text(json.dumps(history, indent=2))

Colored Output

Add color codes for different response types:

GREEN = class="hl-str">"\033[92m"
YELLOW = class="hl-str">"\033[93m"
RED = class="hl-str">"\033[91m"
RESET = class="hl-str">"\033[0m"

def format_state(state):
    colors = {class="hl-str">"completed": GREEN, class="hl-str">"working": YELLOW, class="hl-str">"failed": RED}
    color = colors.get(state, RESET)
    return fclass="hl-str">"{color}{state}{RESET}"

Pipe-Friendly Mode

Detect non-interactive input for scripting:

import sys

if not sys.stdin.isatty():
    class=class="hl-str">"hl-comment"># Non-interactive: read all lines, process each, exit
    for line in sys.stdin:
        result = send_message(headers, line.strip())
        handle_response(headers, result)
else:
    class=class="hl-str">"hl-comment"># Interactive REPL
    ...