async-agent¶

Status: Current | Last reviewed: 2026-10-13

Runs LangGraph workflows. Anything that doesn't need to fit inside a real-time call — analytics, CRM updates, multi-step reasoning, scheduled syncs — flows through here.

Source¶

apps/async-agent

Runtime¶

Aspect	Value
Language	Python 3.12
Framework	LangGraph (StateGraph) on FastAPI
Port	8000
Queue	RabbitMQ in cluster and Vanguard
Workflow job concurrency	Four background slots plus two reserved interactive slots by default
Background workers	Scheduler, provider batch poller, batch flush poller, Salesforce Bulk query poller, durable For Each worker

What it does¶

Receives an event (lifecycle, webhook, scheduled, direct).
Resolves matching rows in trigger_rules (workflow-triggers.md).
For each match, builds a LangGraph StateGraph from workflow_versions.config (langgraph-construction.md).
Executes nodes; persists checkpoints; emits results.

Direct calls with execution_mode="async" atomically persist the execution and a private dispatch envelope in scheduled_jobs, then publish immediately with confirmation. Failed publication remains recoverable. HTTP and agent-host MCP use this same path, with optional tenant/workflow-scoped Idempotency-Key and an admitted root configuration snapshot. The submitted execution_id can be polled at /api/v1/workflow-runs/{id} before a run exists.

Direct calls with execution_mode="sync" also create a workflow_executions row and a workflow_runs row before building the graph, then finalize both rows with outputs or error details. This includes an agent's synchronous lifecycle.on_start hook and in-process synchronous workflow-as-tool calls, so start-of-conversation and nested runs and their per-node traces appear in the workflow's execution history just like manual, triggered, scheduled, and asynchronous runs. Both IDs are returned in the sync HTTP response; they must be full UUIDs because node trace rows reference them as UUID foreign keys.

RabbitMQ uses separate channels, prefetch and semaphores for background (WORKFLOW_JOB_CONCURRENCY=4) and priority >=5 (WORKFLOW_INTERACTIVE_CONCURRENCY=2) work. WORKFLOW_WORKER_LANE selects all, interactive or background; interactive-only processes omit background pollers. Replicas multiply capacity. Provider/model limits still apply.

Database, Redis and durable checkpoint storage warm before startup completes. Execution leases renew with ownership checks, and duplicate deliveries reuse the existing run/checkpoint. New asynchronous runs remain pending with no started_at until the first useful node wrapper is entered. Internal /metrics and run metadata expose startup milestones. See the startup and recovery runbook for measurements, retention requirements and recovery limits.

Node Types¶

Voice flows can overlap their on-start workflow with welcome audio through the authenticated preparation endpoint. Postgres owns a stable call execution; Redis caches its result. The connector joins that execution even after an uncertain acceptance response. See voice startup preparation for its API, revision/caller checks, bounded waits and retry limits.

The registry maps a string node type to a Python builder. Current workflow types include LLM, Salesforce, HTTP request, tool, assignment, router, wait, foreach, parallel, human review, notification, subworkflow, OCR extract, transcribe, score generation, and lifecycle nodes. Full construction details live in langgraph-construction.md.

Outbound notification sends (the notification node and Start Conversation's sms/email link delivery — both go through the shared ChannelRegistry.send) pass the tenant data-protection gate for the notifications boundary (data-protection.md): app/services/data_protection.py resolves the tenant policy (~60 s TTL) and, per its action, sends unchanged (off/no policy), logs a warning (flag), redacts PII from the rendered body/subject (redact), or refuses the send on the node's error path (block). Resolver or detector faults fail open — the message goes out unchanged.

Dependencies¶

Service	Purpose
Postgres	`workflow_versions`, `trigger_rules`, checkpoint tables
Redis	LangGraph short-lived state
RabbitMQ	Event queue (cluster only)
Object storage	Paged workflow datasets (`workflow-datasets/...`)
ai-orchestrator	`agent_call` nodes that invoke an agent
agent-host	`tool_call` nodes
Knowledge Sources API	MCP search endpoint for feature-flagged Knowledge tools
Provider APIs	Per node config

Shared Python libraries are loaded from the repo/image at runtime. In particular, app/http_client_bootstrap.py adds libs/http-client before workflow tool and node modules load, because lazy MCP/OCR/transcription paths import http_client.* directly for SSRF-safe outbound calls. MCP tools with config.auth.type="integration" resolve credentials at call time; Salesforce-backed MCP tools mint a bearer token from the linked integration before connecting to Salesforce Platform MCP. The current non-secret config.client_id is authoritative; an encrypted client_id left by an older integration shape is used only when the config field is absent.

app/tools/db_loader.py registers tool_configs rows into the shared ToolRegistry (exposed to agents via the auto-MCP server at /mcp). It loads type_id IN ('http', 'python_sandboxed', 'mcp', 'sfdc_action', 'sfdc', 'openapi'), plus knowledge when the run's tenant has Knowledge enabled and fully configured. Knowledge rows are registered as MCP-backed search tools: the runtime resolves the tenant's Knowledge settings (tenant_settings key knowledge) through the shared libs/knowledge resolver, derives {base_url}/mcp/{source_slug}, injects the decrypted Bearer token, and does not expose any other method. There is no env fallback. The API tool type (type_id='openapi') carries two subtypes via config.subtype: only manual (a single hand-authored endpoint) executes natively here — through _make_manual_api_handler, which templates {var} placeholders from the LLM args then session context, resolves auth from inline tool_secrets or the linked integration (_resolve_api_auth), SSRF-checks the URL, and applies an optional response_path. The openapi (spec) subtype is skipped on this path — it's discovered/handed to CES, not a single HTTP call.

create_default_registry() also registers the canonical built-in inject_message_to_conversation. Workflow Tool nodes call it by that unprefixed name; it posts to conversation-api with the destination conversation ID and supports both live voice and chat conversations. The migration from the former configured Inject Message to Voice / Inject Message to Conversation rows rewrites workflow references to this built-in before deleting those shared rows.

The auto-MCP registry is a mounted FastMCP streamable-HTTP app at /mcp with the stream path rooted at /. The parent FastAPI lifespan enters the mounted server's session manager before serving requests and exits it during shutdown; without that lifespan bridge, saved MCP tools can appear registered but fail when the streamable session manager has not started.

Workflow-local inline tools are registered into the same ToolRegistry from workflow_config.inline_tools before graph construction. Tool nodes reference them as inline:<id>, and every Tool node in the workflow can call any local definition. Inline Python tools reuse the code-executor sandbox handler, with workflow context snapshots (vars, node_results, trigger event, iterator) forwarded so migrated Code-node sources can keep execute(ctx). Inline API tools reuse the manual API handler for a single configured endpoint. Inline Knowledge tools expose their local workflow tool name, then call the fixed Knowledge MCP search method at {tenant base_url}/mcp/{source_slug} with the tenant's Knowledge token; they are skipped when the run's tenant has Knowledge disabled or incompletely configured.

Tool input self-healing¶

ToolRegistry.execute_tool() applies the same ToolInputIssue envelope used by live agent pipelines. Before a tool handler runs, required fields, scalar types, enum values, and simple patterns are checked from input_schema. After the handler runs, recoverable input errors are classified from explicit needs_input envelopes or 400/422-style payloads that name a caller-fixable field. The envelope is returned as a normal tool result with status: "needs_input", issue, fields[], prompt_hint, original_arguments, retry_attempt, and max_attempts.

Workflow Tool nodes suspend when they receive needs_input: the execution status becomes waiting_input, metadata stores tool_input_request_id and the full issue, Redis emits tool_input_requested, and LangGraph is interrupted. Conversation-backed workflows also best-effort inject prompt_hint into the live conversation so the caller can answer in chat or speech. Non-conversation workflows expose the same prompt and fields through the run API/UI.

Resume through POST /api/v1/executions/{execution_id}/tool-input with { "request_id": "...", "values": { ... } } or { "request_id": "...", "text": "..." }. The endpoint validates the pending request, publishes an execution_resume job carrying tool_input_result, and emits tool_input_provided. On resume, the Tool node patches the original arguments, retries the same tool, and clears needs_tool_input once the call succeeds.

LLM execution tiers¶

llm_agent nodes resolve an effective execution tier at runtime. The order is per-node API override, global API override, node config.execution_tier, workflow batch-mode LLM defaults, then standard. Batch-tier LLM nodes submit provider batch work and suspend until the BatchResultPoller resumes the run. Standard and Flex run synchronously inside the node.

Gemini models are executed through langchain-google-genai / ChatGoogleGenerativeAI. Flex requests send service_tier: "flex" through LangChain model_kwargs; Standard omits the tier. When settings.llm_execution.flex_fallback_to_standard_on_capacity_error or an API override enables fallback, a Gemini Flex 429/503 is retried once on Standard and the rest of that LLM node continues on Standard.

Workflow-level settings.rate_limits[] define shared Redis-backed caps for LLM calls. A profile can match provider/model/integration/tier, and the default bucket key includes tenant, provider, model, and integration id, so parallel workflow runs share the same cap. Each LLM provider call holds a Redis max_in_flight lease until it returns or the lease expires; optional requests_per_minute enforces a rolling 60-second window.

Salesforce connection reuse¶

The native Salesforce client reuses one httpx.AsyncClient per event loop for OAuth, standard REST and Bulk requests. app/services/salesforce_http.py owns the bounded pool: 100 maximum connections, 20 maximum idle connections and a 60-second keepalive expiry. This avoids creating a new TCP/TLS connection for each sequential query or write.

Authentication remains request-specific and integrations are resolved on each call. The shared client rejects cookies, has no default bearer or mutable integration headers, and does not follow redirects. API failures still reach the existing caller; pooling adds no automatic request retries. Normal requests use 30 seconds, while Bulk operations supply their timeout on that request. Salesforce record contents are not cached by this change.

The HTTP lifespan closes the current loop's pool during shutdown, including an exceptional lifespan exit. Standalone callers must await close_salesforce_http_client() before closing their loop. Other event loops receive independent clients.

The focused tests use a real local HTTP server to verify connection reuse and request isolation, plus lifecycle/error-path checks:

cd apps/async-agent
.venv/bin/python -m pytest tests/test_salesforce_http_pool.py \
  tests/test_salesforce_sobject_workflow.py tests/test_salesforce_bulk_workflow.py \
  tests/test_mcp_tool_auth.py -q

Salesforce Bulk Query¶

The Salesforce workflow node supports operation: "query" with query_mode: "standard" | "bulk".

Standard queries keep Salesforce's immediate response under the node's result key. The workflow editor parses the SOQL SELECT list and exposes paths such as <node>.result.records[0].FirstName in the End output mapper. Action nodes using Salesforce Platform MCP return their parsed JSON directly, so the equivalent path is <node>.records[0].FirstName; legacy SFDC Action tools keep the older <node>.data.records[0].FirstName envelope. These three prefixes mirror their runtime handlers and must not be normalized to one UI-only shape.

Bulk mode is intentionally a single node mode, not a create/poll/fetch block pattern. The node requires only integration_id and soql in the common inspector. Advanced defaults control poll interval, timeout, page size, and Salesforce job cleanup.

Runtime flow:

The node creates a Salesforce Bulk API query job and writes a salesforce_bulk_query_jobs row.
It interrupts the LangGraph run, leaving the run checkpointed.
SalesforceBulkQueryPoller polls due rows.
On JobComplete, the poller fetches every final CSV result page.
Pages are stored as gzipped NDJSON artifacts and indexed in workflow_dataset_pages; the manifest lives in workflow_datasets.
The poller publishes an execution_resume job with context.salesforce_bulk_results[<node_id>].
The Salesforce node resumes and returns a compact handle:

{
  "type": "workflow_dataset",
  "dataset_id": "...",
  "row_count": 1000000,
  "page_count": 20,
  "columns": ["Id", "Name"]
}

Large row arrays are never written into workflow state.

Durable For Each¶

For Each still runs normal arrays inline. When items_path resolves to a workflow_dataset handle, the node submits a workflow_fanout_runs row, creates one workflow_fanout_items row per dataset row, and suspends the parent run.

WorkflowFanoutWorker claims due items up to the run's max_in_flight, runs the nested body graph with iterator.current_item, and retries failed items with exponential backoff. The parent workflow resumes with a compact summary:

{
  "status": "ok",
  "count": 1000000,
  "successes": 1000000,
  "failures": 0,
  "durable": true,
  "fanout_run_id": "..."
}

Use per-For Each max_in_flight for local run concurrency and settings.rate_limits[] for shared provider caps across workflows.

Execution stream: SSE + WebSocket¶

GET /sse/executions/{id} and WS /ws/executions/{id} are one stream on two transports: both drive the module-level generator in execution_stream.py (XREAD on async-agent:events:{id}; Redis stream ids are the wire event ids, so Last-Event-ID / ?last_event_id= resume identically). WS frames are {"id","event","data"} / {"comment"} with data the same JSON string an EventSource would carry.

The WS endpoint is ticket-gated — POST /api/v1/executions/{id}/stream-ticket mints a single-use 90 s ticket (stream_tickets.py) over the caller's API token + execution ownership; there is deliberately NO ?access_token= support for long-lived bearers. ai-manager's workflow test route mints one after starting an execution and returns a stream descriptor ({transport, url, ticket}); reconnects re-mint via ai-manager's GET /api/workflows/executions/{id}/stream-ticket.

/ws/executions is async-agent's ONLY externally reachable path: the ai-manager host's ingress (HTTPRoute / dev nginx) path-routes it here, so the workflow editor's browser socket is same-origin with the page and no new base URL exists. Set TSF_EVENT_TRANSPORT=ws on ai-manager to switch the editor to it (see env-vars).

Tests: tests/test_execution_stream_ws.py (tickets single-use, 4401/4403, frame shape, resume, SSE byte-parity).

Vanguard¶

overmind restart async-agent

Direct run from UI: ai-manager → Workflows → Run. Trigger rules also fire from conversation-api lifecycle events and inbound webhook handlers.

Knowledge tools require no env vars — the workflow's tenant must have Knowledge enabled with a base URL and API token under ai-manager Settings → Tenant Settings → Knowledge. The token is decrypted and used only server-side when the workflow runtime calls the Knowledge MCP endpoint.

Production¶

Helm chart at charts/async-agent. One Deployment for the API, one for the consumer that pulls from RabbitMQ.

Execution inspection and control routes require a tenant-scoped Bearer tsf_* token with workflows:manage or agents:manage; the service scopes list, detail, event-log, communications, approve, override, tool-input, and SSE access to the token tenant. ASYNC_AGENT_CORS_ALLOWED_ORIGINS controls the credentialed browser origins; wildcard credentialed CORS is intentionally not used.

The chart NetworkPolicy allowlists ingress from ai-manager, ai-orchestrator, conversation-api and agent-host. Egress is still broad in the base chart because the shared template cannot yet express DNS plus the Postgres/Redis/RabbitMQ/service egress matrix without breaking runtime.

Agent evaluation¶

Performance tests use durable jobs, pinned runtime configurations and shared mock state. See agent evaluation and worker operations for this service’s role, admission limits, authenticated stream ownership and recovery.

Studio roles and runtime tokens¶

Saved workflow execution/tests, event ingestion and execution inspection/resume require agents:run. agents:manage implies run access for authors; workflows have no separate permission gate. Personal tokens derive current access from the active tenant membership: Admin gets full tenant access, Builder gets agent/tool authoring and run access, Viewer gets browse/run access. Platform Admin receives every capability in the token's tenant. Removed memberships and inactive users or tenants are rejected.

Legacy roles resolve as Developer → Builder, Operator → Admin and Analyst → Viewer. Legacy pipeline/call read scopes map to agents:run, while workflow management and pipeline write scopes map to agents:manage. Normalizing a read scope repeatedly never grants write access. See API authentication.

Regression check from the repository root: apps/async-agent/.venv/bin/python scripts/audit/test_runtime_access.py.

Named tenant roles are authoritative. Retired direct permission grants, including wildcards, do not elevate Viewer or Builder.

Native transactional workflows¶

Salesforce nodes support Composite transactions with typed request bodies and subresponse validation. Inline subworkflows can serialize short non-suspending graphs by tenant and key. Strict Assignment errors stop validation failures, and Background workflow status checks verified conversation jobs outside the lock. See Takeda native workflows for configuration, limits and readback/retry behavior.

Checkpoint initialization uses a shared lock: simultaneous first broker jobs reuse one live Postgres saver. Setup failure closes its owner before retry. This prevents one initialization from discarding a connection already used by another job.

Workflow Decisions¶

The Decision node retains the stored router type and stable output IDs. Rules use the existing expression runtime; the visual editor stores JSONata comparisons and nested AND/OR groups. AI-assisted mode calls OpenAI Decisions through the node-level tenant integration before evaluating branch rules. Named answers remain available to downstream nodes. Errors and refusals fail by default and can be explicitly routed.

The authenticated POST /api/v1/decisions/test runtime endpoint and POST /api/workflows/decisions/test Studio proxy preview one node without executing its successors. The proxy requires workflow-authoring permission and applies a rate limit; the runtime binds credentials to the authenticated token tenant and rejects a mismatched tenant header. See the Decision guide for configuration and test examples.