Hosted workflow

Existing Postgres support agent

Keep an existing Postgres support app as source of truth while Synapsor governs agent reads, evidence, proposals, approval, and replay.

Problem

A support app already stores tickets, customers, and policy chunks in Postgres, but exposing a raw SQL tool to an agent would let the model choose scope and writes.

Why app glue gets messy

Most teams end up rebuilding tenant filtering, policy retrieval, proposal review, and audit in app code next to their ORM.

How Synapsor helps

Synapsor imports selected Postgres tables as live-read external mappings, generates a workflow/context/capability pair, and stages risky ticket updates as external writeback proposals.

Value sources

Example names and seed ids are developer-defined. SESSION values are set by your backend. ARG values come from the SDK/HTTP call. Handles such as wrp://..., evidence://..., and agent-run://... are returned by Synapsor and should be stored for audit/replay.

Read the value-source guide

Expected outcome

The agent can answer with evidence and propose ticket updates without direct Postgres write access.

Production checks

Postgres stays the source of truth.
Tenant filtering comes from SESSION, not model arguments.
The read capability records external query audit and evidence.
The proposal capability stages a change for trusted worker writeback after approval.

schema.sql

-- Existing Postgres app tables.
CREATE TABLE customers (
  id TEXT PRIMARY KEY,
  tenant_id TEXT NOT NULL,
  name TEXT NOT NULL,
  email TEXT
);

CREATE TABLE tickets (
  id TEXT PRIMARY KEY,
  tenant_id TEXT NOT NULL,
  customer_id TEXT NOT NULL REFERENCES customers(id),
  subject TEXT NOT NULL,
  body TEXT NOT NULL,
  status TEXT NOT NULL,
  resolution_note TEXT,
  updated_at TIMESTAMP DEFAULT now()
);

CREATE TABLE policy_chunks (
  id TEXT PRIMARY KEY,
  tenant_id TEXT NOT NULL,
  title TEXT NOT NULL,
  body TEXT NOT NULL,
  version INT NOT NULL
);

agent.ddl.sql

CREATE AGENT WORKFLOW support.ticket_resolution_flow
SESSION REQUIRE tenant_id, principal, current_ticket_id
ALLOWED CAPABILITIES (
  support.answer_ticket_question,
  support.propose_ticket_resolution
)
EVIDENCE REQUIRED
AUTO BRANCH ON PROPOSAL
CHECKPOINT EVERY STEP
ON RISK medium REQUIRE APPROVAL ROLE 'support_lead';

CREATE AGENT CONTEXT support.ticket_context
ROOT EXTERNAL external_support.tickets AS ticket
LOOKUP ticket.id = SESSION current_ticket_id
BIND tenant_id FROM SESSION tenant_id
BIND principal FROM SESSION principal
JOIN EXTERNAL external_support.customers AS customer
  ON customer.id = ticket.customer_id
  AND customer.tenant_id = SESSION tenant_id
SEARCH EXTERNAL external_support.policy_chunks AS policy_hits
  USING TEXT(policy_chunks.body)
  QUERY ARG question
  FILTER policy_chunks.tenant_id = SESSION tenant_id
  TOP 5
OUTPUT SLOTS
  ticket_id AS ticket.id,
  subject AS ticket.subject,
  status AS ticket.status,
  customer_name AS customer.name,
  policy_hits AS policy_hits
EVIDENCE ON;

CREATE AGENT CAPABILITY support.answer_ticket_question
ARG question TEXT REQUIRED
HIDDEN tenant_id FROM SESSION tenant_id
HIDDEN principal FROM SESSION principal
HIDDEN current_ticket_id FROM SESSION current_ticket_id
USE CONTEXT support.ticket_context
EXECUTION READ ONLY
REQUIRES EVIDENCE
RETURN ANSWER WITH CITATIONS;

CREATE AGENT CAPABILITY support.propose_ticket_resolution
ARG ticket_id TEXT REQUIRED
ARG new_status TEXT REQUIRED
ARG resolution_note TEXT REQUIRED
HIDDEN tenant_id FROM SESSION tenant_id
HIDDEN principal FROM SESSION principal
USE CONTEXT support.ticket_context
EXECUTION PROPOSAL
REQUIRES EVIDENCE
WRITE PROPOSAL TARGET EXTERNAL external_support.tickets
OPERATION UPDATE
LOOKUP id FROM ARG ticket_id
TENANT tenant_id FROM BINDING tenant_id
COLUMNS status FROM ARG new_status,
        resolution_note FROM ARG resolution_note
AUDIT support.ticket_audit;

python

db.set_session({
    "tenant_id": "acme",
    "principal": "support_agent_17",
    "current_ticket_id": "T001",
})

run = db.agent_runs.start(
    workflow="support.ticket_resolution_flow",
    input={"ticket_id": "T001"},
)

answer = run.invoke_capability(
    "support.answer_ticket_question",
    step_key="answer_ticket",
    arguments={"question": "Can we close this ticket?"},
    response_envelope=True,
)

proposal = run.invoke_capability(
    "support.propose_ticket_resolution",
    step_key="propose_resolution",
    mode="propose_only",
    auto_branch=True,
    arguments={"ticket_id": "T001", "new_status": "resolved", "resolution_note": "Policy evidence attached."},
    response_envelope=True,
)