Connect Email to Truplexy

app.post("/inbound-email", express.json(), async (req, res) => {
  res.sendStatus(200);
  const { From, Subject, TextBody, MessageID } = req.body;   // Postmark's fields
  const { reply } = await ask(`mail:${From}`, `${Subject}\n\n${TextBody}`.slice(0, 4000), { contacts: [{ type: "email", value: From }] });
  if (reply) await sendEmail({ to: From, subject: `Re: ${Subject}`, text: reply, inReplyTo: MessageID });
});

// truplexy.js — runs on your server only.
// TRUPLEXY_CHAT_KEY lives in your secret manager, never in a browser or app.
const API = "https://api.truplexy.com/v2";

export class TruplexyApiError extends Error {
  constructor(status, error, requestId) {
    super(error.message ?? "Truplexy request failed");
    this.status = status; this.code = error.code ?? "UPSTREAM_ERROR";
    this.requestId = error.request_id ?? requestId; this.details = error;
  }
}
export async function call(path, body, method = "POST") {
  const res = await fetch(API + path, {
    method,
    headers: { Authorization: `Bearer ${process.env.TRUPLEXY_CHAT_KEY}`, "Content-Type": "application/json" },
    body: body ? JSON.stringify(body) : undefined,
  });
  const data = await res.json().catch(() => ({}));
  if (!res.ok) throw new TruplexyApiError(res.status, data.error ?? {}, res.headers.get("X-Request-ID"));
  return data;
}

export async function upsertCustomer(customer) {
  try { return await call("/customer", customer, "PUT"); }
  catch (e) {
    if (e.status !== 409 || e.code !== "CUSTOMER_EXISTS") throw e;
    return call("/customer", customer, "PUT"); // one retry only
  }
}
export async function createConversation(customer, channelId) {
  const known = customer == null ? null : await upsertCustomer(customer);
  if (customer != null && (typeof known?.id !== "string" || !known.id)) throw new TruplexyApiError(502, { code: "UPSTREAM_ERROR", message: "Missing customer ID" });
  return call("/conversations", { ...(known && { customer_id: known.id }), ...(channelId && { channel_id: channelId }) });
}

// One Truplexy conversation per customer thread. Use your database in production.
const threads = new Map();
export const threadFor = (conversationId) => [...threads].find(([, c]) => c === conversationId)?.[0];

export async function ask(thread, text, customer, channelId) {
  let id = threads.get(thread);
  if (!id) threads.set(thread, (id = (await createConversation(customer, channelId)).id));
  const r = await call(`/conversations/${id}/messages`, { message: text.slice(0, 4000) });
  // "escalated" (a person has it) and "context" have no reply: send nothing.
  return { reply: r.reply, status: r.status, conversationId: id };
}

export const handoff = (id) => call(`/conversations/${id}/handoff`);

from fastapi import BackgroundTasks, FastAPI, Request, Response
from truplexy import ask

app = FastAPI()


@app.post("/inbound-email")
async def inbound_email(request: Request, background: BackgroundTasks):
    background.add_task(handle_email, await request.json())
    return Response(status_code=200)


async def handle_email(mail: dict):   # Postmark's fields
    r = await ask(f"mail:{mail['From']}", f"{mail['Subject']}\n\n{mail['TextBody']}"[:4000], {"contacts": [{"type": "email", "value": mail["From"]}]})
    if r["reply"]:
        await send_email(to=mail["From"], subject=f"Re: {mail['Subject']}", text=r["reply"], in_reply_to=mail["MessageID"])

# truplexy.py — runs on your server only.
# TRUPLEXY_CHAT_KEY lives in your secret manager, never in a browser or app.
import os
import httpx

API = "https://api.truplexy.com/v2"
_http = httpx.AsyncClient(
    base_url=API,
    headers={"Authorization": f"Bearer {os.environ['TRUPLEXY_CHAT_KEY']}"},
    timeout=60,
)


class TruplexyApiError(Exception):
    def __init__(self, response, error):
        super().__init__(error.get("message", "Truplexy request failed"))
        self.status, self.code = response.status_code, error.get("code", "UPSTREAM_ERROR")
        self.request_id = error.get("request_id") or response.headers.get("x-request-id")
        self.details = error


async def call(path: str, body: dict | None = None, method: str = "POST") -> dict:
    res = await _http.request(method, path, **({"json": body} if body is not None else {}))
    try:
        data = res.json()
    except ValueError:
        data = {}
    if res.is_error:
        raise TruplexyApiError(res, data.get("error", {}))
    return data


async def upsert_customer(customer: dict) -> dict:
    try:
        return await call("/customer", customer, "PUT")
    except TruplexyApiError as e:
        if e.status != 409 or e.code != "CUSTOMER_EXISTS":
            raise
        return await call("/customer", customer, "PUT")


async def create_conversation(customer=None, channel_id=None) -> dict:
    opening = {"channel_id": channel_id} if channel_id else {}
    if customer is not None:
        opening["customer_id"] = (await upsert_customer(customer))["id"]
    return await call("/conversations", opening)


# One Truplexy conversation per customer thread. Use your database in production.
threads: dict[str, str] = {}


def thread_for(conversation_id: str) -> str | None:
    return next((t for t, c in threads.items() if c == conversation_id), None)


async def ask(thread: str, text: str, customer=None, channel_id=None) -> dict:
    conv = threads.get(thread)
    if not conv:
        conv = threads[thread] = (await create_conversation(customer, channel_id))["id"]
    r = await call(f"/conversations/{conv}/messages", {"message": text[:4000]})
    # "escalated" (a person has it) and "context" have no reply: send nothing.
    return {"reply": r.get("reply"), "status": r["status"], "conversation_id": conv}


async def handoff(conversation_id: str) -> dict:
    return await call(f"/conversations/{conversation_id}/handoff")

<?php // inbound-email.php — Postmark's inbound webhook
require __DIR__ . '/truplexy.php';

$mail = json_decode(file_get_contents('php://input'), true);
http_response_code(200);
if (function_exists('fastcgi_finish_request')) fastcgi_finish_request();   // answer at once, keep working

$r = ask('mail:' . $mail['From'], "{$mail['Subject']}\n\n{$mail['TextBody']}", ['contacts' => [['type' => 'email', 'value' => $mail['From']]]]);
if ($r['reply']) send_email($mail['From'], 'Re: ' . $mail['Subject'], $r['reply'], $mail['MessageID']);

<?php // truplexy.php — runs on your server only.
// TRUPLEXY_CHAT_KEY lives in your secret manager, never in a browser or app.
const TRUPLEXY_API = 'https://api.truplexy.com/v2';

class TruplexyApiError extends RuntimeException {
    public function __construct(public int $status, public array $details, public string $requestId = '') {
        parent::__construct($details['message'] ?? 'Truplexy request failed');
    }
    public function apiCode(): string { return $this->details['code'] ?? 'UPSTREAM_ERROR'; }
}

/** Calls an API with a JSON body and returns the decoded answer. */
function post_json(string $url, ?array $body = null, array $headers = [], string $method = 'POST'): array {
    $ch = curl_init($url);
    curl_setopt_array($ch, [
        CURLOPT_CUSTOMREQUEST => $method,
        CURLOPT_RETURNTRANSFER => true,
        CURLOPT_TIMEOUT => 60,
        CURLOPT_HTTPHEADER => array_merge(['Content-Type: application/json'], $headers),
    ]);
    if ($method !== 'GET') curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($body ?? new stdClass()));
    $requestId = '';
    curl_setopt($ch, CURLOPT_HEADERFUNCTION, function ($ch, $line) use (&$requestId) {
        if (stripos($line, 'X-Request-ID:') === 0) $requestId = trim(substr($line, 13));
        return strlen($line);
    });
    $res = curl_exec($ch);
    $status = curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
    $data = json_decode((string) $res, true) ?? [];
    if ($res === false || $status >= 300) throw new TruplexyApiError($status ?: 502, $data['error'] ?? [], $data['error']['request_id'] ?? $requestId);
    return $data;
}

function truplexy(string $path, ?array $body = null, string $method = 'POST'): array {
    return post_json(TRUPLEXY_API . $path, $body, ['Authorization: Bearer ' . getenv('TRUPLEXY_CHAT_KEY')], $method);
}

function upsert_customer(array $customer): array {
    try { return truplexy('/customer', $customer, 'PUT'); }
    catch (TruplexyApiError $e) {
        if ($e->status !== 409 || $e->apiCode() !== 'CUSTOMER_EXISTS') throw $e;
        return truplexy('/customer', $customer, 'PUT');
    }
}
function create_conversation(?array $customer = null, ?string $channelId = null): array {
    $opening = $channelId ? ['channel_id' => $channelId] : [];
    if ($customer !== null) {
        $known = upsert_customer($customer);
        if (empty($known['id'])) throw new TruplexyApiError(502, ['code' => 'UPSTREAM_ERROR', 'message' => 'Missing customer ID']);
        $opening['customer_id'] = $known['id'];
    }
    return truplexy('/conversations', $opening ?: null);
}

// One Truplexy conversation per customer thread. Use your database in production.
function threads(?array $save = null): array {
    $file = __DIR__ . '/threads.json';
    if ($save !== null) file_put_contents($file, json_encode($save), LOCK_EX);
    return $save ?? (json_decode(@file_get_contents($file) ?: '{}', true));
}

function thread_for(string $conversationId): ?string {
    return array_search($conversationId, threads(), true) ?: null;
}

function ask(string $thread, string $text, ?array $customer = null, ?string $channelId = null): array {
    $threads = threads();
    $id = $threads[$thread] ??= create_conversation($customer, $channelId)['id'];
    threads($threads);
    $r = truplexy("/conversations/$id/messages", ['message' => mb_substr($text, 0, 4000)]);
    // "escalated" (a person has it) and "context" have no reply: send nothing.
    return ['reply' => $r['reply'], 'status' => $r['status'], 'conversationId' => $id];
}

function handoff(string $conversationId): array {
    return truplexy("/conversations/$conversationId/handoff");
}

http.HandleFunc("POST /inbound-email", func(w http.ResponseWriter, r *http.Request) {
	var mail struct{ From, Subject, TextBody, MessageID string }   // Postmark's fields
	json.NewDecoder(r.Body).Decode(&mail)
	w.WriteHeader(http.StatusOK)
	go func() {
		a, err := ask("mail:"+mail.From, mail.Subject+"\n\n"+mail.TextBody, identify("email", mail.From))
		if err == nil && a.Reply != "" {
			sendEmail(mail.From, "Re: "+mail.Subject, a.Reply, mail.MessageID)
		}
	}()
})

// truplexy.go — runs on your server only.
// TRUPLEXY_CHAT_KEY lives in your secret manager, never in a browser or app.
package main

import (
	"bytes"
	"encoding/json"
	"io"
	"net/http"
	"os"
	"sync"
	"time"
)

const truplexyAPI = "https://api.truplexy.com/v2"

var httpClient = &http.Client{Timeout: 60 * time.Second}

type APIError struct {
    Status int
    Code, Message, RequestID string
    Details map[string]any
}
func (e *APIError) Error() string { return e.Code + ": " + e.Message }

// sendJSON calls an API with a JSON body, decoding the answer into out when given.
func sendJSON(method, url string, body any, headers map[string]string, out ...any) error {
	var payload io.Reader = http.NoBody
	if body != nil {
		b, err := json.Marshal(body)
		if err != nil {
			return err
		}
		payload = bytes.NewReader(b)
	}
	req, err := http.NewRequest(method, url, payload)
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	for k, v := range headers {
		req.Header.Set(k, v)
	}
	res, err := httpClient.Do(req)
	if err != nil {
		return err
	}
	defer res.Body.Close()
	if res.StatusCode >= 300 {
		msg, _ := io.ReadAll(res.Body)
        var data struct { Error map[string]any }
        json.Unmarshal(msg, &data)
        code, _ := data.Error["code"].(string)
        message, _ := data.Error["message"].(string)
        requestID, _ := data.Error["request_id"].(string)
        if requestID == "" { requestID = res.Header.Get("X-Request-ID") }
        return &APIError{res.StatusCode, code, message, requestID, data.Error}
	}
	if len(out) == 0 {
		return nil
	}
	return json.NewDecoder(res.Body).Decode(out[0])
}

func call(path string, body, out any) error { return callMethod("POST", path, body, out) }
func callMethod(method, path string, body, out any) error {
	auth := map[string]string{"Authorization": "Bearer " + os.Getenv("TRUPLEXY_CHAT_KEY")}
	if out == nil {
		return sendJSON(method, truplexyAPI+path, body, auth)
	}
	return sendJSON(method, truplexyAPI+path, body, auth, out)
}

func upsertCustomer(customer map[string]any) (string, error) {
    var result struct { ID string }
    err := callMethod("PUT", "/customer", customer, &result)
    if e, ok := err.(*APIError); ok && e.Status == 409 && e.Code == "CUSTOMER_EXISTS" {
        err = callMethod("PUT", "/customer", customer, &result)
    }
    if err == nil && result.ID == "" { err = &APIError{Status: 502, Code: "UPSTREAM_ERROR", Message: "Missing customer ID"} }
    return result.ID, err
}
type ConversationOptions struct { Customer map[string]any; ChannelID string }
func identify(kind, value string) ConversationOptions {
    return ConversationOptions{Customer: map[string]any{"contacts": []map[string]string{{"type": kind, "value": value}}}}
}
func createConversation(options ConversationOptions) (string, error) {
    opening := map[string]string{}
    if options.ChannelID != "" { opening["channel_id"] = options.ChannelID }
    if options.Customer != nil {
        id, err := upsertCustomer(options.Customer)
        if err != nil { return "", err }
        opening["customer_id"] = id
    }
    var result struct { ID string }
    err := call("/conversations", opening, &result)
    return result.ID, err
}

// One Truplexy conversation per customer thread. Use your database in production.
var (
	mu      sync.Mutex
	threads = map[string]string{}
)

func threadFor(conversationID string) string {
	mu.Lock()
	defer mu.Unlock()
	for t, c := range threads {
		if c == conversationID {
			return t
		}
	}
	return ""
}

type Answer struct {
	Reply          string `json:"reply"` // empty for "escalated" and "context": send nothing
	Status         string `json:"status"`
	ConversationID string `json:"conversationId"`
}

func ask(thread, text string, options ...ConversationOptions) (Answer, error) {
	mu.Lock()
	id, ok := threads[thread]
	mu.Unlock()
	if !ok {
        option := ConversationOptions{}
        if len(options) > 0 { option = options[0] }
        var err error
        id, err = createConversation(option)
        if err != nil { return Answer{}, err }
		mu.Lock()
		threads[thread] = id
		mu.Unlock()
	}
	if r := []rune(text); len(r) > 4000 {
		text = string(r[:4000])
	}
	var a Answer
	err := call("/conversations/"+id+"/messages", map[string]string{"message": text}, &a)
	a.ConversationID = id
	return a, err
}

func handoff(conversationID string) error {
	return call("/conversations/"+conversationID+"/handoff", map[string]any{}, nil)
}

// Your team acted in Truplexy → act on it here.
app.post("/truplexy-webhook", express.raw({ type: "application/json" }), async (req, res) => {
  const raw = req.body.toString("utf8");
  if (!verify(raw, req.get("X-Truplexy-Signature") ?? "", process.env.TRUPLEXY_WEBHOOK_SECRET)) return res.sendStatus(401);
  res.sendStatus(200);                                   // answer within 2.5 seconds
  const { type, data } = JSON.parse(raw);
  const conversationId = data.conversation_id;
  if (type === "message.created") {                      // your team replied
    const text = data.message.content;
    const agent = data.message.agent ?? "Support";
    const thread = threadFor(conversationId);
    await sendEmail({ to: thread.replace("mail:", ""), subject: "Re: your request", text });
  }
});

import crypto from "node:crypto";

export function verify(rawBody, header, secret) {
  const { t, v1 } = Object.fromEntries(header.split(",").map((part) => part.split("=")));
  const expected = crypto.createHmac("sha256", secret).update(`${t}.${rawBody}`).digest("hex");
  const fresh = Math.abs(Date.now() / 1000 - Number(t)) < 300;
  return fresh && v1?.length === expected.length && crypto.timingSafeEqual(Buffer.from(v1), Buffer.from(expected));
}

import json
import os
from fastapi import BackgroundTasks, Request, Response
from truplexy import thread_for
from verify import verify


# Your team acted in Truplexy → act on it here.
@app.post("/truplexy-webhook")
async def truplexy_webhook(request: Request, background: BackgroundTasks):
    raw = await request.body()
    if not verify(raw, request.headers.get("X-Truplexy-Signature", ""), os.environ["TRUPLEXY_WEBHOOK_SECRET"]):
        return Response(status_code=401)
    background.add_task(on_event, json.loads(raw))   # answer within 2.5 seconds
    return Response(status_code=200)


async def on_event(event: dict):
    data = event["data"]
    conversation_id = data.get("conversation_id")
    if event["type"] == "message.created":   # your team replied
        text = data["message"]["content"]
        agent = data["message"].get("agent") or "Support"
        thread = thread_for(conversation_id)
        await send_email(to=thread.removeprefix("mail:"), subject="Re: your request", text=text)

import hashlib
import hmac
import time


def verify(raw_body: bytes, header: str, secret: str) -> bool:
    parts = dict(p.split("=", 1) for p in header.split(",") if "=" in p)
    t, v1 = parts.get("t", ""), parts.get("v1", "")
    expected = hmac.new(secret.encode(), t.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    fresh = t.isdigit() and abs(time.time() - int(t)) < 300
    return fresh and hmac.compare_digest(v1, expected)

<?php // webhook.php — your team acted in Truplexy → act on it here.
require __DIR__ . '/truplexy.php';
require __DIR__ . '/verify.php';

$raw = file_get_contents('php://input');
if (!verify_truplexy($raw, $_SERVER['HTTP_X_TRUPLEXY_SIGNATURE'] ?? '', getenv('TRUPLEXY_WEBHOOK_SECRET'))) {
    http_response_code(401);
    exit;
}
http_response_code(200);
if (function_exists('fastcgi_finish_request')) fastcgi_finish_request();   // answer at once, keep working

$event = json_decode($raw, true);
$data = $event['data'];
$conversationId = $data['conversation_id'] ?? null;
if ($event['type'] === 'message.created') {   // your team replied
    $text = $data['message']['content'];
    $agent = $data['message']['agent'] ?? 'Support';
    $thread = thread_for($conversationId);
    send_email(substr($thread, 5), 'Re: your request', $text);
}

<?php // verify.php
function verify_truplexy(string $rawBody, string $header, string $secret): bool {
    $parts = [];
    foreach (explode(',', $header) as $part) {
        [$k, $v] = explode('=', $part, 2) + [1 => ''];
        $parts[$k] = $v;
    }
    $t = $parts['t'] ?? '';
    $expected = hash_hmac('sha256', "$t.$rawBody", $secret);
    return ctype_digit($t) && abs(time() - (int) $t) < 300 && hash_equals($expected, $parts['v1'] ?? '');
}

// Your team replied in Truplexy → deliver it on the platform.
http.HandleFunc("POST /truplexy-webhook", func(w http.ResponseWriter, r *http.Request) {
	raw, _ := io.ReadAll(r.Body)
	if !verify(raw, r.Header.Get("X-Truplexy-Signature"), os.Getenv("TRUPLEXY_WEBHOOK_SECRET")) {
		w.WriteHeader(http.StatusUnauthorized)
		return
	}
	w.WriteHeader(http.StatusOK)   // answer within 2.5 seconds
	var event struct {
		Type string `json:"type"`
		Data struct {
			ConversationID string `json:"conversation_id"`
			Message        struct{ Content, Agent string }
		}
	}
	if json.Unmarshal(raw, &event) != nil || event.Type != "message.created" {
		return
	}
	agent := event.Data.Message.Agent
	if agent == "" {
		agent = "Support"
	}
	go deliver(threadFor(event.Data.ConversationID), event.Data.Message.Content, agent)
})

func deliver(thread, text, agent string) {
	sendEmail(strings.TrimPrefix(thread, "mail:"), "Re: your request", text, "")
}

package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"strconv"
	"strings"
	"time"
)

func verify(rawBody []byte, header, secret string) bool {
	var t, v1 string
	for _, part := range strings.Split(header, ",") {
		k, v, _ := strings.Cut(part, "=")
		switch k {
		case "t":
			t = v
		case "v1":
			v1 = v
		}
	}
	ts, err := strconv.ParseInt(t, 10, 64)
	if err != nil || max(time.Now().Unix()-ts, ts-time.Now().Unix()) >= 300 {
		return false
	}
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(t + "."))
	mac.Write(rawBody)
	return hmac.Equal([]byte(hex.EncodeToString(mac.Sum(nil))), []byte(v1))
}