Knowledge for agents

import os, requests

API = "https://api.60db.ai"
H = {"Authorization": f"Bearer {os.environ['SIXTYDB_API_KEY']}"}

r = requests.post(f"{API}/memory/ingest", headers=H, timeout=30, json={
    "text": "Open 9 to 5, Monday to Saturday. Closed on Sundays.",
    "title": "Opening hours",
    "type": "knowledge",
    "collection": "bistro_faq",
})
r.raise_for_status()

r = requests.patch(f"{API}/agents/{agent_id}", headers=H, timeout=30,
                   json={"knowledge_collection_id": collection_uuid})
r.raise_for_status()

import hashlib, hmac, json, os, time
from http.server import BaseHTTPRequestHandler, HTTPServer

SECRET = os.environ["SIXTYDB_WEBHOOK_SECRET"].encode()  # whsec_...
TOLERANCE_SECONDS = 300


def verify(raw: bytes, timestamp: str, signature: str) -> bool:
    try:
        if abs(time.time() - int(timestamp)) > TOLERANCE_SECONDS:
            return False
    except (TypeError, ValueError):
        return False
    expected = "v1=" + hmac.new(SECRET, timestamp.encode() + b"." + raw, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature or "")


def search(query: str, max_results: int) -> list[str]:
    return ["Open 9 to 5, Monday to Saturday."][:max_results]  # your retrieval here


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        raw = self.rfile.read(int(self.headers.get("Content-Length", 0)))
        if not verify(raw, self.headers.get("X-60db-Timestamp"), self.headers.get("X-60db-Signature")):
            self.send_response(401); self.end_headers(); return
        try:
            body = json.loads(raw)
            passages = search(str(body["query"]), int(body.get("max_results", 4)))
        except (ValueError, KeyError):
            self.send_response(400); self.end_headers(); return
        out = json.dumps({"passages": passages}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(out)))
        self.end_headers()
        self.wfile.write(out)


HTTPServer(("", 8080), Handler).serve_forever()

package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"io"
	"log"
	"math"
	"net/http"
	"os"
	"strconv"
	"time"
)

var secret = []byte(os.Getenv("SIXTYDB_WEBHOOK_SECRET")) // whsec_...

const toleranceSeconds = 300

func verify(raw []byte, timestamp, signature string) bool {
	ts, err := strconv.ParseInt(timestamp, 10, 64)
	if err != nil || math.Abs(float64(time.Now().Unix()-ts)) > toleranceSeconds {
		return false
	}
	mac := hmac.New(sha256.New, secret)
	mac.Write([]byte(timestamp + "."))
	mac.Write(raw)
	expected := "v1=" + hex.EncodeToString(mac.Sum(nil))
	return hmac.Equal([]byte(expected), []byte(signature))
}

func search(query string, max int) []string {
	return []string{"Open 9 to 5, Monday to Saturday."} // your retrieval here
}

func handler(w http.ResponseWriter, r *http.Request) {
	raw, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 1<<16))
	if err != nil {
		http.Error(w, "bad request", http.StatusBadRequest)
		return
	}
	if !verify(raw, r.Header.Get("X-60db-Timestamp"), r.Header.Get("X-60db-Signature")) {
		http.Error(w, "unauthorized", http.StatusUnauthorized)
		return
	}
	var body struct {
		Query      string `json:"query"`
		MaxResults int    `json:"max_results"`
	}
	if err := json.Unmarshal(raw, &body); err != nil {
		http.Error(w, "bad request", http.StatusBadRequest)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(map[string]any{"passages": search(body.Query, body.MaxResults)})
}

func main() {
	http.HandleFunc("/search", handler)
	log.Fatal(http.ListenAndServe(":8080", nil))
}