Create sourcing workflow

Outcome

Save a scheduled workflow that keeps a list topped up: every interval (hourly, daily or weekly) it adds up to maxPerRun (default 100, at most 1000) active contacts matching a saved segment that are not yet on the list. It is active and first runs at the next scheduled tick (within five minutes). source.kind 'crm' is refused with 409 crm_source_needs_service_access: TopoloCRM does not yet accept a service credential, so a scheduled job cannot read it; use contacts.import_from_crm by hand instead. The write is audited. Errors: 404 workflow_not_found, list or segment not found; 409 archived list.

Before you start

• Verify the intended environment, signed-in credential, authorized organization and the action permission workspace:write.
• Discover and select an authorized canonical Engage workspace. Pass resourceType=workspace and resourceId separately from action input.
• Review the exact effects and obtain approval before CLI --confirm or MCP confirm:true. A local capture does not authorize a live mutation.

Discover the target

topolo whoami --json
topolo actions get app_topolo_engage.sourcing.workflows.create --json
topolo workspaces list topolo-engage --json

Set ENGAGE_WORKSPACE_ID to the identifier of the selected authorized workspace.

Input contract

Validate, plan and execute

Build payload.json from the current schema and discovered identifiers. The action description above defines its behavior, limits, side effects and declared errors.

: "${ENGAGE_WORKSPACE_ID:?Set ENGAGE_WORKSPACE_ID to your selected workspace identifier}"
topolo actions validate app_topolo_engage.sourcing.workflows.create --data @payload.json --json
topolo actions plan app_topolo_engage.sourcing.workflows.create --resource-type workspace --resource-id "$ENGAGE_WORKSPACE_ID" --data @payload.json --confirm --json

Review the plan and approve its exact effects before executing the unchanged request.

: "${ENGAGE_WORKSPACE_ID:?Set ENGAGE_WORKSPACE_ID to your selected workspace identifier}"
topolo actions call app_topolo_engage.sourcing.workflows.create --resource-type workspace --resource-id "$ENGAGE_WORKSPACE_ID" --data @payload.json --confirm --json

For MCP, use topolo_get_action, topolo_validate_action, topolo_plan_action and topolo_call_action. Pass resource context separately where required; confirm:true is required for mutations.

Captured successful requests and responses

Captured HTTP 201 · sourcing workflows over the published actions > pauses itself when its list is archived or its segment deleted, and records the failure

Action input

{
  "name": "Acme top-up",
  "source": {
    "kind": "segment",
    "segmentId": "segment_1"
  },
  "listId": "list_1",
  "interval": "daily",
  "maxPerRun": 2
}

Actual local response

{
  "workflow": {
    "id": "sourcingWorkflow_1",
    "name": "Acme top-up",
    "status": "active",
    "pausedReason": null,
    "source": {
      "kind": "segment",
      "segmentId": "segment_1"
    },
    "listId": "list_1",
    "interval": "daily",
    "maxPerRun": 2,
    "nextRunAt": "2026-10-01T09:00:00.000Z",
    "lastRunAt": null,
    "lastRunSummary": null,
    "createdBy": "user_1",
    "createdAt": "2026-10-01T09:00:00.000Z",
    "updatedAt": "2026-10-01T09:00:00.000Z"
  }
}

Source test: sourcing-routes.test.ts. Actual route handlers with SQLite repositories; synthetic identity/workspace grants and provider ports.

Captured HTTP 201 · sourcing workflows over the published actions > creates, lists, gets, updates and deletes a workflow with audited writes

Action input

{
  "name": "Default max",
  "source": {
    "kind": "segment",
    "segmentId": "segment_1"
  },
  "listId": "list_1",
  "interval": "daily"
}

Actual local response

{
  "workflow": {
    "id": "sourcingWorkflow_2",
    "name": "Default max",
    "status": "active",
    "pausedReason": null,
    "source": {
      "kind": "segment",
      "segmentId": "segment_1"
    },
    "listId": "list_1",
    "interval": "daily",
    "maxPerRun": 100,
    "nextRunAt": "2026-10-01T09:00:00.000Z",
    "lastRunAt": null,
    "lastRunSummary": null,
    "createdBy": "user_1",
    "createdAt": "2026-10-01T09:00:00.000Z",
    "updatedAt": "2026-10-01T09:00:00.000Z"
  }
}

Source test: sourcing-routes.test.ts. Actual route handlers with SQLite repositories; synthetic identity/workspace grants and provider ports.

Captured HTTP 201 · sourcing workflows over the published actions > pauses itself when its list is archived or its segment deleted, and records the failure

Action input

{
  "name": "Acme top-up",
  "source": {
    "kind": "segment",
    "segmentId": "segment_1"
  },
  "listId": "list_2",
  "interval": "daily",
  "maxPerRun": 2
}

Actual local response

{
  "workflow": {
    "id": "sourcingWorkflow_2",
    "name": "Acme top-up",
    "status": "active",
    "pausedReason": null,
    "source": {
      "kind": "segment",
      "segmentId": "segment_1"
    },
    "listId": "list_2",
    "interval": "daily",
    "maxPerRun": 2,
    "nextRunAt": "2026-10-01T09:00:00.000Z",
    "lastRunAt": null,
    "lastRunSummary": null,
    "createdBy": "user_1",
    "createdAt": "2026-10-01T09:00:00.000Z",
    "updatedAt": "2026-10-01T09:00:00.000Z"
  }
}

Source test: sourcing-routes.test.ts. Actual route handlers with SQLite repositories; synthetic identity/workspace grants and provider ports.

Effects and response meaning

• Save a scheduled workflow that keeps a list topped up: every interval (hourly, daily or weekly) it adds up to maxPerRun (default 100, at most 1000) active contacts matching a saved segment that are not yet on the list. It is active and first runs at the next scheduled tick (within five minutes). source.kind 'crm' is refused with 409 crm_source_needs_service_access: TopoloCRM does not yet accept a service credential, so a scheduled job cannot read it; use contacts.import_from_crm by hand instead. The write is audited. Errors: 404 workflow_not_found, list or segment not found; 409 archived list.

Verify the outcome

• Check the response against the declared output schema and the requested operation. For mutations, use the action-specific read described in the action description before any dependent operation.

Recover

For an uncertain result, inspect current retained state with a read action before retrying. Do not replay a mutation unless its contract explicitly guarantees idempotency.

Refused requests

Captured HTTP 404 · sourcing workflows over the published actions > refuses missing or archived lists, missing segments and a TopoloCRM source

Action input

{
  "name": "Acme top-up",
  "source": {
    "kind": "segment",
    "segmentId": "segment_1"
  },
  "listId": "list_9",
  "interval": "daily",
  "maxPerRun": 2
}

Actual local response

{
  "error": {
    "code": "not_found",
    "message": "List was not found in this workspace."
  }
}

Source test: sourcing-routes.test.ts. Actual route handlers with SQLite repositories; synthetic identity/workspace grants and provider ports.

Captured HTTP 404 · sourcing workflows over the published actions > refuses missing or archived lists, missing segments and a TopoloCRM source

Action input

{
  "name": "Acme top-up",
  "source": {
    "kind": "segment",
    "segmentId": "segment_9"
  },
  "listId": "list_1",
  "interval": "daily",
  "maxPerRun": 2
}

Actual local response

{
  "error": {
    "code": "segment_not_found",
    "message": "Segment was not found in this workspace."
  }
}

Source test: sourcing-routes.test.ts. Actual route handlers with SQLite repositories; synthetic identity/workspace grants and provider ports.

Captured HTTP 409 · sourcing workflows over the published actions > refuses missing or archived lists, missing segments and a TopoloCRM source

Action input

{
  "name": "Acme top-up",
  "source": {
    "kind": "crm",
    "q": "acme",
    "updatedAfter": "2026-09-01T00:00:00.000Z"
  },
  "listId": "list_1",
  "interval": "daily",
  "maxPerRun": 2
}

Actual local response

{
  "error": {
    "code": "crm_source_needs_service_access",
    "message": "A scheduled workflow cannot read TopoloCRM yet: TopoloCRM accepts only a person's own session or API key, not a service credential. Import from TopoloCRM by hand with contacts.import_from_crm."
  }
}

Source test: sourcing-routes.test.ts. Actual route handlers with SQLite repositories; synthetic identity/workspace grants and provider ports.

Captured HTTP 422 · sourcing workflows over the published actions > refuses missing or archived lists, missing segments and a TopoloCRM source

Refused input; do not execute

{
  "name": " Acme top-up ",
  "source": {
    "kind": "segment",
    "segmentId": "segment_1"
  },
  "listId": "list_1",
  "interval": "daily",
  "maxPerRun": 1001
}

Actual local response

{
  "error": {
    "code": "invalid_action_input",
    "message": "Input for sourcing.workflows.create is invalid.",
    "details": [
      {
        "origin": "number",
        "code": "too_big",
        "maximum": 1000,
        "inclusive": true,
        "path": [
          "maxPerRun"
        ],
        "message": "Too big: expected number to be <=1000"
      }
    ]
  }
}

Source test: sourcing-routes.test.ts. Actual route handlers with SQLite repositories; synthetic identity/workspace grants and provider ports.

Captured HTTP 409 · sourcing workflows over the published actions > refuses missing or archived lists, missing segments and a TopoloCRM source

Action input

{
  "name": "Acme top-up",
  "source": {
    "kind": "segment",
    "segmentId": "segment_1"
  },
  "listId": "list_2",
  "interval": "daily",
  "maxPerRun": 2
}

Actual local response

{
  "error": {
    "code": "state_conflict",
    "message": "Archived lists cannot be sourced into."
  }
}

Source test: sourcing-routes.test.ts. Actual route handlers with SQLite repositories; synthetic identity/workspace grants and provider ports.

Captured HTTP 403 · sourcing workflows over the published actions > lets readers read but not write

Action input

{
  "name": "Acme top-up",
  "source": {
    "kind": "segment",
    "segmentId": "segment_1"
  },
  "listId": "list_1",
  "interval": "daily",
  "maxPerRun": 2
}

Actual local response

{
  "error": {
    "code": "permission_denied",
    "message": "The credential lacks workspace:write.",
    "details": {
      "requiredPermission": "workspace:write"
    }
  }
}

Source test: sourcing-routes.test.ts. Actual route handlers with SQLite repositories; synthetic identity/workspace grants and provider ports.

Captured HTTP 422 · sourcing workflows over the published actions > refuses missing or archived lists, missing segments and a TopoloCRM source

Refused input; do not execute

{
  "name": " Acme top-up ",
  "source": {
    "kind": "segment",
    "segmentId": "segment_1"
  },
  "listId": "list_1",
  "interval": "monthly",
  "maxPerRun": 2
}

Actual local response

{
  "error": {
    "code": "invalid_action_input",
    "message": "Input for sourcing.workflows.create is invalid.",
    "details": [
      {
        "code": "invalid_value",
        "values": [
          "hourly",
          "daily",
          "weekly"
        ],
        "path": [
          "interval"
        ],
        "message": "Invalid option: expected one of \"hourly\"|\"daily\"|\"weekly\""
      }
    ]
  }
}

Source test: sourcing-routes.test.ts. Actual route handlers with SQLite repositories; synthetic identity/workspace grants and provider ports.

Error recovery

• HTTP 404 not_found: Rediscover the identifier using the related list or get action in the same authorized resource. Do not guess identifiers or change organization context.
• HTTP 404 segment_not_found: Rediscover the identifier using the related list or get action in the same authorized resource. Do not guess identifiers or change organization context.
• HTTP 409 crm_source_needs_service_access: Read the refusal details and current action contract, correct the stated cause, then validate and plan again. Do not retry unchanged or infer success.
• HTTP 422 invalid_action_input: Correct the request against the current action input schema, then validate it again before planning.
• HTTP 409 state_conflict: Read the current retained state and reconcile the requested transition before retrying. Do not blindly replay a non-idempotent mutation.
• HTTP 403 permission_denied: Stop and inspect the current credential and selected resource. It needs workspace:write; do not broaden access or switch organization automatically.

Related actions

• List sourcing workflows
• Get sourcing workflow
• Update sourcing workflow
• Pause sourcing workflow
• Resume sourcing workflow
• Delete sourcing workflow
• Run sourcing workflow now
• List sourcing runs

Evidence and limits

• /api/sourcing/workflows

• sourcing-routes.test.ts

• Actual route handlers with SQLite repositories; synthetic identity/workspace grants and provider ports.

• Widget/bootstrap use actual app handler but mocked worker runtime and app identity.

• No gateway execution, live permissions, external delivery or deployed proof.

These are local route captures with synthetic identity and simulated dependencies. They do not prove live authorization, gateway execution, production state, or external delivery.