agent_tool_endpoint.py

"""An endpoint for a Subcontext agent tool, in Python 3.9 or later with the standard library only.

It offers two operations you can add to a custom integration on Subcontext:

  GET  /policies/{{ reference }}   look up a policy (the agent sends date_of_birth too)
  POST /callbacks                  book a callback (the body comes from your body template)

Subcontext sends your API key in the X-API-Key header (the integration's
Auth type "Api key"). Writes carry an Idempotency-Key header that stays the
same if the same tool call is sent again, so this file answers a repeat with
the first answer instead of booking twice.

Run it:
  SUBCONTEXT_TOOL_KEY=A_LONG_RANDOM_VALUE python3 agent_tool_endpoint.py
Subcontext only calls https addresses on the public internet, so put it
behind your usual HTTPS web server or load balancer.
"""

import hmac
import json
import os
import re
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, unquote, urlparse

# The value you store as a secret variable on the integration in Subcontext.
API_KEY = os.environ.get("SUBCONTEXT_TOOL_KEY")
PORT = int(os.environ.get("PORT", "4100"))
if not API_KEY:
    raise SystemExit("Set SUBCONTEXT_TOOL_KEY to the API key Subcontext sends.")

# Stand-ins for your own systems.
POLICIES = {
    "POL-1001": {"date_of_birth": "1980-04-12", "status": "active",
                 "renewal_date": "2027-03-01", "monthly_premium": "18.40"},
}
ANSWERED_WRITES = {}  # Idempotency-Key -> (status, body)


class Handler(BaseHTTPRequestHandler):
    def send_json(self, status, body):
        data = json.dumps(body).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def key_matches(self):
        given = self.headers.get("X-API-Key", "")
        return hmac.compare_digest(given.encode(), API_KEY.encode())

    def do_GET(self):
        # 1. Check the key. A 401 tells Subcontext the credentials are wrong.
        if not self.key_matches():
            return self.send_json(401, {"error": "Unknown API key."})

        # 2. A read: GET /policies/POL-1001?date_of_birth=1980-04-12
        url = urlparse(self.path)
        match = re.fullmatch(r"/policies/([^/]+)", url.path)
        if not match:
            return self.send_json(404, {"error": "No such operation."})

        reference = unquote(match.group(1)).upper()
        date_of_birth = parse_qs(url.query).get("date_of_birth", [""])[0]
        policy = POLICIES.get(reference)
        # Ask for something only the customer knows before giving anything away.
        if not policy or policy["date_of_birth"] != date_of_birth:
            # The agent sees this status and message, so make the message useful.
            return self.send_json(404, {"error": "No policy matches that reference and date of birth. "
                                                 "Ask the customer to check both."})
        # Small, flat and only what the agent needs to answer.
        self.send_json(200, {"reference": reference, "status": policy["status"],
                             "renewal_date": policy["renewal_date"],
                             "monthly_premium": policy["monthly_premium"]})

    def do_POST(self):
        if not self.key_matches():
            return self.send_json(401, {"error": "Unknown API key."})
        if urlparse(self.path).path != "/callbacks":
            return self.send_json(404, {"error": "No such operation."})

        # 3. A write: POST /callbacks with the body your template builds.
        idempotency_key = self.headers.get("Idempotency-Key")
        if idempotency_key and idempotency_key in ANSWERED_WRITES:
            return self.send_json(*ANSWERED_WRITES[idempotency_key])

        length = int(self.headers.get("Content-Length") or 0)
        if length > 100_000:
            return self.send_json(413, {"error": "The request body is too large."})
        try:
            body = json.loads(self.rfile.read(length) or b"{}")
        except ValueError:
            return self.send_json(400, {"error": "The request body wasn't valid JSON."})
        if not body.get("phone"):
            return self.send_json(422, {"error": "A phone number is needed to book a callback. "
                                                 "Ask the customer for one."})

        answer = (201, {"callback_id": str(uuid.uuid4()),
                        "message": "Callback booked for "
                                   f"{body.get('preferred_time') or 'the next available slot'}."})
        if idempotency_key:
            ANSWERED_WRITES[idempotency_key] = answer
        self.send_json(*answer)


if __name__ == "__main__":
    print(f"Agent tool endpoint on http://localhost:{PORT}")
    ThreadingHTTPServer(("", PORT), Handler).serve_forever()