Pre-execution verification for LLM-generated agentic workflows.
LLM-generated workflows are powerful but dangerous. When an AI agent builds a pipeline that reads from Salesforce, calls Clearbit, and writes back to your CRM, you're trusting that it got the types right, declared its side effects honestly, and won't silently corrupt your data. Today there's no verification layer between "the LLM generated a workflow" and "the workflow is running in production." workflow-verify closes that gap.
workflow-verify defines a Workflow AST (Abstract Syntax Tree) that LLMs generate instead of raw code. This AST is then verified for correctness before any code executes:
┌─────────────┐
Prompt ──────────>│ LLM Agent │
└──────┬──────┘
│ Workflow AST (JSON)
v
┌─────────────┐
│ Verify │──── Type flow
│ Engine │──── Schema validity
│ │──── Side effects
│ │──── Guard conditions
└──────┬──────┘
│ VerificationResult
┌────┴────┐
│ │
✅ Pass ❌ Fail
│ │
v v
┌────────┐ ┌──────────┐
│Transpile│ │Self-Correct│
│to Code │ │& Retry │
└────────┘ └──────────┘
│
v
TypeScript / Python / Temporal
pip install workflow-verifyfrom workflow_verify import Workflow, verify, transpile, TranspileTarget
import json
# Load a workflow (typically generated by an LLM)
workflow = Workflow(**json.load(open("workflow.json")))
# Verify before execution
result = verify(workflow)
print(result.trace)
if result.passed:
# Safe to transpile and run
code = transpile(workflow, TranspileTarget.PYTHON)
print(code.code)
else:
for error in result.errors:
print(f" {error.message}")
print(f" Fix: {error.suggestion}")Every step's output type must be compatible with the next step's input type. workflow-verify catches mismatches before they become runtime errors.
Step 1: output → EnrichedLead (has: email, name, company, company_size, industry, website)
Step 2: input → ScoredLead (needs: email, name, company, company_size, industry, website, score)
^^^^^ missing!
❌ Step 'score_leads' input field 'score' (Int) not found in previous output 'EnrichedLead'.
FIX: Add field 'score' to schema 'EnrichedLead', or remove it from 'ScoredLead'.
Supported types: Text, Int, Float, Bool, Email, URL, Phone, Date, DateTime, Json, Any, plus List[T], Optional[T], and structural Record types. Subtyping rules are built in (e.g., Email is a subtype of Text, Int is a subtype of Float).
Schemas must be defined before they're referenced, names must be unique, and fields must have valid types.
Every step must declare what it reads, writes, calls, sends, or deletes. Undeclared effects are caught:
⚠️ Step 'enrich_contact' description suggests a 'call' effect but none is declared.
FIX: Add an effect { "kind": "call", "target": "clearbit" } to this step.
Guards reference fields that must exist in the step's input schema:
❌ Guard 'score >= 70' in step 'push_qualified' references field 'score',
but 'score' is not in the input schema.
FIX: Available fields: email, name, company.
Every verification produces a human-readable audit trail:
$ wfv verify workflow.json
✅ Schema 'RawLead' — 3 fields validated.
✅ Schema 'EnrichedLead' — 6 fields validated.
✅ Schema 'ScoredLead' — 7 fields validated.
✅ Schema 'PushResult' — 3 fields validated.
✅ Step 'fetch_leads' input compatible with workflow input 'RawLead'.
✅ Step 'push_qualified' output satisfies workflow output 'PushResult'.
✅ Step 'enrich_leads' input compatible with 'fetch_leads' output.
✅ Step 'score_leads' input compatible with 'enrich_leads' output.
✅ Step 'push_qualified' input compatible with 'score_leads' output.
⚠️ Step 'fetch_leads' description/config suggests a 'send' effect but none is declared.
✅ Effects manifest: READ:salesforce, CALL:clearbit, WRITE:salesforce.
✅ Step 'push_qualified' guard 'score >= 70' references valid field 'score'.
Effects (3):
read:salesforce
call:clearbit
write:salesforce
Verification passed.
# Verify a workflow
wfv verify workflow.json
wfv verify workflow.json --json # Machine-readable output
wfv verify workflow.json --no-strict # Warnings don't fail
# Transpile to code
wfv transpile workflow.json -t python
wfv transpile workflow.json -t typescript
wfv transpile workflow.json -t temporal -o output.py
# Browse the schema registry
wfv registry list # All schemas
wfv registry list crm # Filter by category
wfv registry search lead # Search by keyword
wfv registry show crm/salesforce_lead # Show schema details
# Generate from prompt (requires LLM API key)
wfv generate "Fetch leads from Salesforce, enrich with Clearbit, score, push back" -t pythonVerified workflows can be transpiled to production-ready code:
| Target | Output | Dependencies |
|---|---|---|
python |
Pydantic models + async pipeline | pydantic |
typescript |
Zod schemas + typed functions | zod |
temporal |
Temporal.io workflow + activities | temporalio |
from workflow_verify import transpile, TranspileTarget
result = transpile(workflow, TranspileTarget.TYPESCRIPT)
print(result.code) # Generated TypeScript
print(result.filename) # Suggested filename
print(result.dependencies) # ["zod"]
print(result.instructions) # Setup instructions20 pre-built schemas across 5 categories, ready to use in your workflows:
| Category | Schemas |
|---|---|
| crm | Salesforce Lead/Contact/Opportunity, HubSpot Contact/Deal, CRMZero Contact |
| enrichment | Clearbit Person/Company, Clay Enrichment, Apollo Contact |
| communication | Slack Message, Email Message, Webhook Payload |
| data | Postgres Record, Stripe Customer, CSV Row |
| common | Person, Company, Address, Money |
from workflow_verify import load_schema, search_schemas, list_categories
# Load a specific schema
lead = load_schema("crm/salesforce_lead")
print(lead.name, "—", len(lead.fields), "fields")
# Search across all schemas
results = search_schemas("email")
for schema in results:
print(f"{schema.name}: {schema.description}")
# List categories
print(list_categories()) # ['common', 'communication', 'crm', 'data', 'enrichment']For live schemas from production APIs:
from workflow_verify import resolve_schema
# Fetches the actual schema from your HubSpot instance
schema = await resolve_schema(
"hubspot", "contacts",
credentials={"access_token": "pat-xxx"},
include_custom=True, # Include custom properties
fallback_to_static=True, # Fall back to registry if API fails
)Supported services: HubSpot, Salesforce, Stripe, PostgreSQL, Clay, CRMZero.
When paired with an LLM, workflow-verify drives an automatic correction loop: generate, verify, fix errors, re-verify — until the workflow passes or attempts are exhausted.
from workflow_verify import run_sync
# One-liner: prompt → verified, transpiled code
code = run_sync(
"Build a pipeline that fetches leads from Salesforce, "
"enriches them with Clearbit, scores them, and pushes "
"qualified leads back to Salesforce",
target="python",
)
print(code)Or with full control:
from workflow_verify import generate_and_verify
result = await generate_and_verify(
prompt="...",
target="typescript",
llm="anthropic", # or "openai"
max_attempts=3,
)
if result.converged:
print(f"Converged in {len(result.attempts)} attempt(s)")
print(result.transpiled.code)
else:
print("Failed to converge")
for attempt in result.attempts:
if attempt.verification:
print(f" Attempt {attempt.attempt_number}: {len(attempt.verification.errors)} errors")Requires an LLM provider: pip install workflow-verify[llm]
workflow-verify ships as an MCP (Model Context Protocol) server, letting LLMs like Claude verify workflows inline during conversations.
pip install workflow-verify[mcp]Two tools are exposed:
| Tool | Description |
|---|---|
verify_workflow |
Takes a workflow AST JSON, verifies it, optionally transpiles, returns result + trace |
generate_verified_workflow |
Takes a prompt, runs the full generate-verify-correct loop, returns AST + code |
Claude Desktop configuration (~/Library/Application Support/Claude/claude_desktop_config.json):
{
"mcpServers": {
"workflow-verify": {
"command": "uv",
"args": ["--directory", "/path/to/workflow_verify",
"run", "python", "-m", "workflow_verify.mcp_server"]
}
}
}Once configured, Claude can verify and transpile workflows directly in conversation.
All symbols are available from the top-level package:
import workflow_verifyWorkflow, Step, Schema, FieldDef, Effect, Guard
WFType, ListType, OptionalType, RecordType, RecordField, AnyWFType, is_compatible()
verify(), CheckResult, VerificationResult
transpile(), TranspileTarget, TranspileResult
generate_and_verify(), run(), run_sync(), effects(), LLMClient, Attempt, CorrectionRequest, CorrectionResult, format_correction_request()
get_workflow_json_schema(), get_workflow_tool_definition(), load_schema(), list_schemas(), list_categories(), search_schemas(), resolve_schema(), SchemaLoadError
format_trace()
# Core (verification + transpilation + registry)
pip install workflow-verify
# With LLM support (Anthropic + OpenAI)
pip install workflow-verify[llm]
# With Temporal transpiler runtime
pip install workflow-verify[temporal]
# MCP server for Claude integration
pip install workflow-verify[mcp]
# Development
pip install workflow-verify[dev]See CONTRIBUTING.md for guidelines, especially for adding new schemas to the registry.
MIT