Implementing Tool Backends

Implementing Tool Backends

This guide shows how to build secure tool endpoints that integrate with OnceOnly.

Overview

When you register a tool in OnceOnly, you’re pointing to an endpoint that handles the actual work. This endpoint must:

1. Verify HMAC signatures - Ensure requests come from OnceOnly
2. Handle the payload - Extract arguments and execute
3. Return results - Send JSON response
4. Handle errors - Return proper HTTP status codes

FastAPI Example

Here’s a complete FastAPI implementation with two tools:

Setup

pip install fastapi uvicorn

Complete Implementation

from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib
import json
import time

app = FastAPI()

# 🔐 Use the same secret you specified when creating the tool in OnceOnly
TOOL_SECRET = "supersecret"

# ---- HMAC verification ----
def verify_signature(raw_body: bytes, headers: dict):
    """Verify that request came from OnceOnly"""
    sig = headers.get("x-onceonly-signature")
    ts = headers.get("x-onceonly-timestamp")
    alg = headers.get("x-onceonly-signature-alg")

    if not sig or not ts:
        raise HTTPException(status_code=401, detail="Missing signature headers")
    if alg and alg.lower() != "hmac_sha256":
        raise HTTPException(status_code=401, detail="Unsupported signature algorithm")

    # (Optional) Replay attack protection
    if abs(time.time() - int(ts)) > 300:
        raise HTTPException(status_code=401, detail="Timestamp expired")

    expected = hmac.new(
        TOOL_SECRET.encode(),
        raw_body,
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(expected, sig):
        raise HTTPException(status_code=401, detail="Invalid signature")

# =========================================================
# ✅ TOOL 1 — SUCCESS CASE
# =========================================================
@app.post("/tools/send-email")
async def send_email_tool(request: Request):
    """Send email via third-party service"""
    raw = await request.body()
    verify_signature(raw, request.headers)

    payload = json.loads(raw)
    args = payload.get("args", {})

    to = args.get("to")
    subject = args.get("subject")
    body = args.get("body")

    # 🟢 This is where you perform the ACTUAL action
    print(f"[SEND EMAIL] to={to} subject={subject}")

    # Call your email service (SendGrid, AWS SES, etc)
    # email_service.send(to, subject, body)

    return {
        "status": "sent",
        "to": to,
        "subject": subject,
        "timestamp": time.time()
    }

# =========================================================
# ❌ TOOL 2 — INTENTIONAL ERROR
# =========================================================
@app.post("/tools/fail-test")
async def fail_tool(request: Request):
    """Test endpoint that simulates failure"""
    raw = await request.body()
    verify_signature(raw, request.headers)

    # Simulating a third-party API crash/failure
    raise HTTPException(status_code=500, detail="Simulated tool failure")

# =========================================================
# ✅ TOOL 3 — WITH IDEMPOTENCY
# =========================================================

# Store processed requests to implement idempotency on tool side too
processed = {}

@app.post("/tools/create-ticket")
async def create_ticket_tool(request: Request):
    """Create support ticket with idempotency"""
    raw = await request.body()
    verify_signature(raw, request.headers)

    payload = json.loads(raw)
    args = payload.get("args", {})
    lease_id = payload.get("lease_id")
    # Optional: accept a custom idempotency key passed in args
    idempotency_key = args.get("idempotency_key") or lease_id

    # Check if already processed
    if idempotency_key and idempotency_key in processed:
        print(f"[DUPLICATE] Returning cached result for key={idempotency_key}")
        return processed[idempotency_key]

    # Process new request
    title = args.get("title")
    description = args.get("description")
    priority = args.get("priority", "normal")

    print(f"[CREATE TICKET] title={title} priority={priority}")

    # Create ticket in your system
    ticket = {
        "id": f"TKT-{int(time.time())}",
        "title": title,
        "description": description,
        "priority": priority,
        "created_at": time.time()
    }

    # Cache result
    if idempotency_key:
        processed[idempotency_key] = {
            "status": "created",
            "ticket": ticket
        }
        return processed[idempotency_key]

    return {
        "status": "created",
        "ticket": ticket
    }

# =========================================================
# ✅ TOOL 4 — BATCH OPERATION
# =========================================================
@app.post("/tools/send-bulk-emails")
async def send_bulk_emails_tool(request: Request):
    """Send emails to multiple recipients"""
    raw = await request.body()
    verify_signature(raw, request.headers)

    payload = json.loads(raw)
    args = payload.get("args", {})

    recipients = args.get("recipients", [])
    template = args.get("template")
    data = args.get("data", {})

    print(f"[BULK EMAIL] Sending {len(recipients)} emails")

    results = []
    for email in recipients:
        try:
            # Send email to each recipient
            # email_service.send_template(email, template, data)
            results.append({
                "email": email,
                "status": "sent"
            })
        except Exception as e:
            results.append({
                "email": email,
                "status": "failed",
                "error": str(e)
            })

    sent = sum(1 for r in results if r["status"] == "sent")
    failed = sum(1 for r in results if r["status"] == "failed")

    return {
        "status": "completed",
        "sent": sent,
        "failed": failed,
        "results": results
    }

# =========================================================
# HEALTH CHECK
# =========================================================
@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "ok"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Run the Server

python app.py
# Server running at http://localhost:8000

Registering Tools in OnceOnly

After your endpoints are running, register them in OnceOnly:

import os
from onceonly import OnceOnly

client = OnceOnly(api_key=os.environ["ONCEONLY_API_KEY"])

# Register send-email tool
client.gov.create_tool({
    "name": "send_email",
    "scope_id": "global",
    "url": "https://your-domain.com/tools/send-email",
    "auth": {
        "type": "hmac_sha256",
        "secret": "supersecret",  # Same as TOOL_SECRET above
    },
    "timeout_ms": 15000,
    "max_retries": 2,
})

# Register create-ticket tool
client.gov.create_tool({
    "name": "create_ticket",
    "scope_id": "global",
    "url": "https://your-domain.com/tools/create-ticket",
    "auth": {
        "type": "hmac_sha256",
        "secret": "supersecret",
    },
    "timeout_ms": 10000,
    "max_retries": 2,
})

Request Format from OnceOnly

When OnceOnly calls your tool, it sends:

POST /tools/send-email HTTP/1.1
Host: your-domain.com
Content-Type: application/json
X-OnceOnly-Signature: <hmac-sha256-signature>
X-OnceOnly-Signature-Alg: hmac_sha256
X-OnceOnly-Timestamp: 1705322400
X-OnceOnly-Tool: send_email
X-OnceOnly-Agent-Id: support_bot
X-OnceOnly-Secret-Id: sec_a1b2c3d4

{
  "tool": "send_email",
  "args": {
    "to": "user@example.com",
    "subject": "Welcome!",
    "body": "Thanks for signing up",
    "idempotency_key": "email_user_123_welcome"
  },
  "agent_id": "support_bot",
  "ns": "user_abc",
  "key_hash": "k_1234abcd",
  "ts": 1705322400,
  "lease_id": "lease_abc123xyz",
  "scope_id": "global"
}

Response Format

Your endpoint should return JSON:

Success:

{
  "status": "sent",
  "to": "user@example.com",
  "subject": "Welcome!",
  "timestamp": 1705322400
}

Error:

{
  "error": "invalid_email",
  "message": "Email address is not valid"
}

Best Practices

✅ DO

1. Always verify signature - Never skip HMAC validation
2. Implement idempotency - Cache results by lease_id or a custom idempotency_key in args
3. Handle retries - Be prepared for duplicate requests
4. Set appropriate timeout - Keep it under 30 seconds
5. Log requests - For debugging integration issues
6. Return consistent format - Always JSON
7. Use proper HTTP status - 200 for success, 4xx for client errors, 5xx for server errors

# Good practice
@app.post("/tools/my-tool")
async def my_tool(request: Request):
    # 1. Log incoming request
    logger.info(f"Received tool call: {request.path}")

    # 2. Verify signature
    raw = await request.body()
    verify_signature(raw, request.headers)

    # 3. Parse and validate
    payload = json.loads(raw)
    validate_payload(payload)

    # 4. Execute with error handling
    try:
        result = execute_tool(payload)
        return result
    except Exception as e:
        logger.error(f"Tool failed: {e}")
        raise HTTPException(status_code=500, detail=str(e))

❌ DON’T

1. Don’t skip signature verification - Security risk
2. Don’t hardcode secrets - Use environment variables
3. Don’t process without idempotency key - Causes duplicates
4. Don’t return sensitive data - Filter secrets from responses
5. Don’t ignore errors - Always return proper status codes
6. Don’t timeout too long - Keep under 30 seconds
7. Don’t modify request body - Use raw bytes for signature

Environment Variables

.env

TOOL_SECRET=supersecret
SENDGRID_API_KEY=sg_xxxxxxxxxxxxx
AWS_SES_ACCESS_KEY=xxx
DATABASE_URL=postgresql://...

import os
from dotenv import load_dotenv

load_dotenv()

TOOL_SECRET = os.getenv("TOOL_SECRET")
SENDGRID_API_KEY = os.getenv("SENDGRID_API_KEY")

Testing Locally

Test with curl

# Generate HMAC signature
BODY='{"tool":"send_email","args":{"to":"test@example.com","subject":"Test","idempotency_key":"test_1"},"agent_id":"support_bot","ns":"user_abc","key_hash":"k_test","ts":1705322400,"lease_id":"lease_test_1","scope_id":"global"}'
SIG=$(echo -n "$BODY" | openssl dgst -sha256 -hmac "supersecret" | cut -d' ' -f2)

# Send request
curl -X POST http://localhost:8000/tools/send-email \
  -H "Content-Type: application/json" \
  -H "X-OnceOnly-Signature: $SIG" \
  -H "X-OnceOnly-Signature-Alg: hmac_sha256" \
  -H "X-OnceOnly-Timestamp: $(date +%s)" \
  -H "X-OnceOnly-Tool: send_email" \
  -H "X-OnceOnly-Agent-Id: support_bot" \
  -d "$BODY"

Test with Python

import requests
import hmac
import hashlib
import json
import time

TOOL_SECRET = "supersecret"
BASE_URL = "http://localhost:8000"

def call_tool(endpoint: str, args: dict, lease_id: str = None):
    """Call tool with proper HMAC signature"""
    if not lease_id:
        lease_id = f"lease_test_{int(time.time())}"

    payload = {
        "tool": "send_email",
        "args": args,
        "agent_id": "support_bot",
        "ns": "user_abc",
        "key_hash": "k_test",
        "ts": int(time.time()),
        "lease_id": lease_id,
        "scope_id": "global"
    }

    body = json.dumps(payload).encode()
    sig = hmac.new(TOOL_SECRET.encode(), body, hashlib.sha256).hexdigest()

    response = requests.post(
        f"{BASE_URL}{endpoint}",
        data=body,
        headers={
            "Content-Type": "application/json",
            "X-OnceOnly-Signature": sig,
            "X-OnceOnly-Signature-Alg": "hmac_sha256",
            "X-OnceOnly-Timestamp": str(int(time.time())),
            "X-OnceOnly-Tool": "send_email",
            "X-OnceOnly-Agent-Id": "support_bot"
        }
    )

    return response.json()

# Test send-email
result = call_tool("/tools/send-email", {
    "to": "test@example.com",
    "subject": "Test",
    "body": "Test body"
})
print(result)

Deployment

Docker

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY app.py .

EXPOSE 8000

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

Production Checklist

• Verify HMAC signatures
• Use environment variables for secrets
• Implement proper logging
• Add error handling
• Set appropriate timeouts
• Use HTTPS in production
• Monitor health checks
• Rate limit if needed
• Add request ID tracing
• Test with OnceOnly signatures

---

Related: Tools Registry | Idempotent Tools