#!/usr/bin/env python3 """ Minimal LangGraph-style renewal agent + Check Write before Salesforce update. Modes (Try → Build → Protect): sandbox — local mock decisions, no key developer — free tc_dev_ key via /developers/get-key → public /api/v1/check-write production — TC Protect™ production key (enterprise controls) """ from __future__ import annotations import json import os import sys from dataclasses import dataclass from typing import Any, Dict try: from dotenv import load_dotenv load_dotenv() except ImportError: pass try: import requests except ImportError: print("Install deps: pip install -r requirements.txt", file=sys.stderr) raise API_URL = os.getenv( "TEKCAPITOL_API_URL", "https://tekcapitol.com/api/v1/check-write", ) API_KEY = os.getenv("TEKCAPITOL_API_KEY", "").strip() _raw_mode = os.getenv("CHECK_WRITE_MODE", "").strip().lower() if _raw_mode in {"sandbox", "developer", "production", "live"}: MODE = "production" if _raw_mode == "live" else _raw_mode elif API_KEY: MODE = "developer" if API_KEY.startswith("tc_dev_") else "production" else: MODE = "sandbox" @dataclass class WriteIntent: agent: str system: str action: str resource: str change: str authority: str context: Dict[str, Any] def to_check_write_body(self) -> Dict[str, Any]: return { "system": self.system, "objectType": "Opportunity", "objectRef": self.resource, "field": "StageName", "agentValue": self.change.replace("Stage = ", "").replace("Stage=", "").strip(), "authoritySource": self.authority, "authorityFound": bool(self.authority) and self.authority.lower() not in {"unknown", "none", "missing", ""}, "metadata": { "agent": self.agent, "requestedChange": self.change, "context": self.context, }, } def sandbox_check_write(intent: WriteIntent) -> Dict[str, Any]: """Deterministic local stand-in labeled as sandbox policy.""" action_l = f"{intent.action} {intent.change}".lower() auth_l = (intent.authority or "").lower() if not intent.authority or auth_l in {"unknown", "none", "missing", "stale"}: return { "decision": "pause", "reason": "Current authority could not be confirmed", "decision_id": "cw_sandbox_local_pause", "allowed": False, "sandbox": True, } if "delete" in action_l or "wipe" in action_l: return { "decision": "block", "reason": "Requested action exceeds sandbox policy", "decision_id": "cw_sandbox_local_block", "allowed": False, "sandbox": True, } signed = intent.context.get("contract_signed") is True if "closed won" in action_l and signed and "sales_ops" in auth_l: return { "decision": "allow", "reason": "Authority and contract context match sandbox policy", "decision_id": "cw_sandbox_local_allow", "allowed": True, "sandbox": True, } return { "decision": "pause", "reason": "Sandbox policy could not allow this write", "decision_id": "cw_sandbox_local_default_pause", "allowed": False, "sandbox": True, } def api_check_write(intent: WriteIntent) -> Dict[str, Any]: r = requests.post( API_URL, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {API_KEY}", }, json=intent.to_check_write_body(), timeout=20, ) r.raise_for_status() data = r.json() data["sandbox"] = False return data def check_write(intent: WriteIntent) -> Dict[str, Any]: if MODE in {"developer", "production"} and API_KEY: return api_check_write(intent) return sandbox_check_write(intent) _SF_STORE: Dict[str, Dict[str, Any]] = {} def update_salesforce_opportunity_stage(record_id: str, stage: str) -> Dict[str, Any]: _SF_STORE[record_id] = {"Id": record_id, "StageName": stage} return _SF_STORE[record_id] def surface_review_state(decision: Dict[str, Any]) -> None: print( "REVIEW REQUIRED (PAUSE):", decision.get("reason"), decision.get("decision_id") or decision.get("auditId"), ) def stop_action(decision: Dict[str, Any]) -> None: print( "STOPPED (BLOCK):", decision.get("reason"), decision.get("decision_id") or decision.get("auditId"), ) def renewal_write_node(state: Dict[str, Any]) -> Dict[str, Any]: """ LangGraph-style node: build Write Intent → Check Write → write only on ALLOW. """ intent = WriteIntent( agent=state.get("agent", "renewal-agent"), system="Salesforce", action="update_opportunity_stage", resource=state["opportunity_id"], change=f"Stage = {state['target_stage']}", authority=state.get("authority", "sales_ops"), context=state.get("context") or {}, ) decision = check_write(intent) d = str(decision.get("decision", "")).lower() print( "Check Write →", d.upper(), "|", decision.get("reason"), "|", decision.get("decision_id") or decision.get("auditId"), ) if d == "allow": updated = update_salesforce_opportunity_stage(state["opportunity_id"], state["target_stage"]) return {**state, "decision": decision, "salesforce": updated, "status": "written"} if d == "pause": surface_review_state(decision) return {**state, "decision": decision, "status": "paused"} stop_action(decision) return {**state, "decision": decision, "status": "blocked"} def run_demo() -> None: print(f"Mode: {MODE} ({'API key set' if API_KEY else 'no key; sandbox mock'})") print(f"Endpoint: {API_URL if MODE != 'sandbox' else '(local sandbox)'}") print("---") cases = [ { "name": "ALLOW path", "opportunity_id": "006ALLOW1", "target_stage": "Closed Won", "authority": "sales_ops", "context": {"contract_signed": True}, }, { "name": "PAUSE path", "opportunity_id": "006PAUSE1", "target_stage": "Closed Won", "authority": "unknown", "context": {"contract_signed": True}, }, { "name": "BLOCK path", "opportunity_id": "006BLOCK1", "target_stage": "Closed Won", "authority": "sales_ops", "context": {"contract_signed": True}, "action_override": "delete_opportunity", }, ] for case in cases: print("\n##", case["name"]) state = { "agent": "renewal-agent", "opportunity_id": case["opportunity_id"], "target_stage": case["target_stage"], "authority": case["authority"], "context": case["context"], } if case.get("action_override"): # Force block via action string for sandbox; API uses engine policy. intent = WriteIntent( agent="renewal-agent", system="Salesforce", action=case["action_override"], resource=case["opportunity_id"], change=f"Stage = {case['target_stage']}", authority=case["authority"], context=case["context"], ) decision = check_write(intent) d = str(decision.get("decision", "")).lower() print("Check Write →", d.upper(), "|", decision.get("reason")) if d == "allow": update_salesforce_opportunity_stage(case["opportunity_id"], case["target_stage"]) elif d == "pause": surface_review_state(decision) else: stop_action(decision) print(json.dumps({"status": d, "decision": decision}, indent=2)[:500]) else: out = renewal_write_node(state) print(json.dumps({"status": out.get("status"), "decision": out.get("decision")}, indent=2)[:500]) if __name__ == "__main__": run_demo()