Protocol Schema
- Schema Viewer
- Schema Source
Loading ....
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "Message",
"description": "The messages supported by the Stepflow protocol (JSON-RPC 2.0).",
"oneOf": [
{
"$ref": "#/$defs/MethodRequest"
},
{
"$ref": "#/$defs/MethodResponse"
},
{
"$ref": "#/$defs/Notification"
}
],
"$defs": {
"ComponentExecuteParams": {
"description": "Sent from Stepflow to the component server to execute a specific component with the provided input.",
"type": "object",
"properties": {
"component": {
"description": "The component to execute.",
"$ref": "#/$defs/Component"
},
"input": {
"description": "The input to the component.",
"$ref": "#/$defs/Value"
},
"attempt": {
"description": "The execution attempt number (1-based).\n\nA monotonically increasing counter that increments on every re-execution\nof this step, regardless of the reason:\n- **Transport error**: The subprocess crashed or a network failure occurred.\n- **Component error**: The component returned an error and the step has\n `onError: { action: retry }`.\n- **Orchestrator recovery**: The orchestrator crashed and is re-executing\n tasks that were in-flight.\n\nComponents can use this to implement idempotency guards or progressive\nfallback strategies.",
"type": "integer",
"format": "uint32",
"minimum": 0
},
"observability": {
"description": "Observability context for tracing and logging.",
"$ref": "#/$defs/ObservabilityContext"
}
},
"required": [
"component",
"input",
"attempt",
"observability"
]
},
"Component": {
"description": "Identifies a specific plugin and atomic functionality to execute. Use component name for builtins (e.g., 'eval') or path format for plugins (e.g., '/python/udf').",
"type": "string",
"examples": [
"/builtin/eval",
"/mcpfs/list_files",
"/python/udf"
]
},
"Value": {
"description": "Any JSON value (object, array, string, number, boolean, or null)"
},
"ObservabilityContext": {
"description": "Observability context for distributed tracing and logging.\n\nThis context is passed with protocol requests to enable trace correlation\nand structured logging across the Stepflow runtime and component servers.\n\n# Field Presence\n\n- `trace_id` and `span_id`: Present when tracing is enabled, None otherwise\n- `run_id` and `flow_id`: Present for workflow execution requests, None for init/discovery\n- `step_id`: Present for step-level execution, None for workflow-level operations",
"type": "object",
"properties": {
"trace_id": {
"description": "OpenTelemetry trace ID (128-bit, hex encoded).\n\nPresent when tracing is enabled, None otherwise.\nUsed to correlate all operations within a single trace.",
"type": [
"string",
"null"
]
},
"span_id": {
"description": "OpenTelemetry span ID (64-bit, hex encoded).\n\nUsed to establish parent-child span relationships.\nComponent servers should use this as the parent span when creating their spans.",
"type": [
"string",
"null"
]
},
"run_id": {
"description": "The ID of the workflow run.\n\nPresent for workflow execution requests, None for initialization/discovery.\nUsed for filtering logs and associating operations with specific workflow runs.",
"type": [
"string",
"null"
]
},
"flow_id": {
"description": "The ID of the flow being executed.\n\nPresent for workflow execution requests, None for initialization/discovery.\nUsed for filtering logs and understanding which workflow is being executed.",
"oneOf": [
{
"$ref": "#/$defs/BlobId"
},
{
"type": "null"
}
]
},
"step_id": {
"description": "The ID of the step being executed.\n\nPresent for step-level execution, None for workflow-level operations.\nUsed for filtering logs and associating operations with specific workflow steps.",
"type": [
"string",
"null"
]
}
}
},
"BlobId": {
"description": "A SHA-256 hash of the blob content, represented as a hexadecimal string.",
"type": "string"
},
"ComponentExecuteResult": {
"description": "Sent from the component server back to Stepflow with the result of the component execution.",
"type": "object",
"properties": {
"output": {
"description": "The result of the component execution.",
"$ref": "#/$defs/Value"
}
},
"required": [
"output"
]
},
"ComponentInfoParams": {
"description": "Sent from Stepflow to the component server to request information about a specific component.",
"type": "object",
"properties": {
"component": {
"description": "The component to get information about.",
"$ref": "#/$defs/Component"
}
},
"required": [
"component"
]
},
"ComponentInfoResult": {
"description": "Sent from the component server back to Stepflow with information about the requested component.",
"type": "object",
"properties": {
"info": {
"description": "Information about the component.",
"$ref": "#/$defs/ComponentInfo"
}
},
"required": [
"info"
]
},
"ComponentInfo": {
"type": "object",
"properties": {
"component": {
"description": "The component ID.",
"$ref": "#/$defs/Component"
},
"description": {
"description": "Optional description of the component.",
"type": [
"string",
"null"
]
},
"input_schema": {
"description": "The input schema for the component.\n\nCan be any valid JSON schema (object, primitive, array, etc.).",
"oneOf": [
{
"$ref": "#/$defs/Schema"
},
{
"type": "null"
}
]
},
"output_schema": {
"description": "The output schema for the component.\n\nCan be any valid JSON schema (object, primitive, array, etc.).",
"oneOf": [
{
"$ref": "#/$defs/Schema"
},
{
"type": "null"
}
]
}
},
"required": [
"component"
]
},
"Schema": {
"description": "A valid JSON Schema object.",
"type": "object",
"additionalProperties": true
},
"ComponentListParams": {
"description": "Sent from Stepflow to the component server to request a list of all available components.",
"type": "object"
},
"ListComponentsResult": {
"description": "Sent from the component server back to Stepflow with a list of all available components.",
"type": "object",
"properties": {
"components": {
"description": "A list of all available components.",
"type": "array",
"items": {
"$ref": "#/$defs/ComponentInfo"
}
}
},
"required": [
"components"
]
},
"ComponentInferSchemaParams": {
"description": "Sent from Stepflow to the component server to infer the output schema for a component\ngiven an input schema. This enables static type checking of workflows.",
"type": "object",
"properties": {
"component": {
"description": "The component to infer the schema for.",
"$ref": "#/$defs/Component"
},
"input_schema": {
"description": "The schema of the input that will be provided to the component.",
"$ref": "#/$defs/Schema"
}
},
"required": [
"component",
"input_schema"
]
},
"ComponentInferSchemaResult": {
"description": "Sent from the component server back to Stepflow with the inferred output schema.",
"type": "object",
"properties": {
"output_schema": {
"description": "The inferred output schema, or None if the component cannot determine it.",
"oneOf": [
{
"$ref": "#/$defs/Schema"
},
{
"type": "null"
}
]
}
}
},
"SubmitRunProtocolParams": {
"description": "Parameters for submitting a run via the protocol.",
"type": "object",
"properties": {
"flowId": {
"description": "The ID of the flow to execute (blob ID).",
"$ref": "#/$defs/BlobId"
},
"inputs": {
"description": "Input values for each item in the run.",
"type": "array",
"items": {
"$ref": "#/$defs/Value"
}
},
"wait": {
"description": "If true, wait for completion before returning.",
"type": "boolean",
"default": false
},
"maxConcurrency": {
"description": "Maximum number of concurrent executions.",
"type": [
"integer",
"null"
],
"format": "uint",
"minimum": 0
},
"overrides": {
"description": "Optional workflow overrides to apply.",
"oneOf": [
{
"$ref": "#/$defs/WorkflowOverrides"
},
{
"type": "null"
}
]
},
"timeoutSecs": {
"description": "Maximum seconds to wait when wait=true (default 300). If the timeout elapses,\nreturns the current run status rather than an error.",
"type": [
"integer",
"null"
],
"format": "uint64",
"minimum": 0
},
"observability": {
"description": "Observability context for tracing.",
"oneOf": [
{
"$ref": "#/$defs/ObservabilityContext"
},
{
"type": "null"
}
]
},
"subflowKey": {
"description": "Client-provided key for subflow deduplication during recovery.\n\nThis key uniquely identifies a subflow submission within the scope\nof the executing step. The orchestrator records this key in the\njournal. If the parent step re-executes after a crash, the\norchestrator matches new submissions by key and returns the existing\nsubflow's run ID instead of creating a duplicate.\n\nThe client must generate the same key on re-execution for recovery\nto work. A common approach is to derive a deterministic UUID from a\nmonotonic counter scoped to the step execution.",
"type": "string",
"format": "uuid"
}
},
"required": [
"flowId",
"inputs",
"subflowKey"
]
},
"WorkflowOverrides": {
"description": "Workflow overrides that can be applied to modify step behavior at runtime.\n\nOverrides are keyed by step ID and contain merge patches or other transformation\nspecifications to modify step properties before execution.",
"type": "object",
"properties": {
"steps": {
"description": "Map of step ID to override specification",
"type": "object",
"additionalProperties": {
"$ref": "#/$defs/StepOverride"
}
}
},
"required": [
"steps"
]
},
"StepOverride": {
"description": "Override specification for a single step.\n\nContains the override type (merge patch, json patch, etc.) and the value\nto apply. The type field uses `$type` to avoid collisions with step properties.",
"type": "object",
"properties": {
"$type": {
"description": "The type of override to apply. Defaults to \"merge_patch\" if not specified.",
"$ref": "#/$defs/OverrideType",
"default": "merge_patch"
},
"value": {
"description": "The override value to apply, interpreted based on the override type."
}
},
"required": [
"value"
]
},
"OverrideType": {
"description": "The type of override operation to perform.\n\nCases:\n* `merge_patch`: Apply a JSON Merge Patch (RFC 7396) to the step.\n\nThis is the default override type. The value should be a JSON object\nwhere null values indicate fields to remove and other values are merged\ninto the target step.\n* `json_patch`: Apply a JSON Patch (RFC 6902) to the step. (Future extension)\n\nThe value should be an array of JSON Patch operations.\nThis is reserved for future use.",
"type": "string",
"enum": [
"merge_patch",
"json_patch"
]
},
"GetRunProtocolParams": {
"description": "Parameters for getting a run via the protocol.",
"type": "object",
"properties": {
"runId": {
"description": "The run ID to query.",
"type": "string"
},
"wait": {
"description": "If true, wait for run completion before returning.",
"type": "boolean",
"default": false
},
"includeResults": {
"description": "If true, include item results in the response.",
"type": "boolean",
"default": false
},
"resultOrder": {
"description": "Order of results (byIndex or byCompletion).",
"$ref": "#/$defs/ResultOrder",
"default": "by_index"
},
"timeoutSecs": {
"description": "Maximum seconds to wait when wait=true (default 300). If the timeout elapses,\nreturns the current run status rather than an error.",
"type": [
"integer",
"null"
],
"format": "uint64",
"minimum": 0
},
"observability": {
"description": "Observability context for tracing.",
"oneOf": [
{
"$ref": "#/$defs/ObservabilityContext"
},
{
"type": "null"
}
]
}
},
"required": [
"runId"
]
},
"ResultOrder": {
"description": "Order in which to return results.\n\nCases:\n* `by_index`: Results in input order (index 0, 1, 2, ...).\n* `by_completion`: Results in completion order (first completed first).",
"type": "string",
"enum": [
"by_index",
"by_completion"
]
},
"RunStatusProtocol": {
"description": "Run status returned by the protocol (serializable version of RunStatus).",
"type": "object",
"properties": {
"runId": {
"type": "string"
},
"flowId": {
"$ref": "#/$defs/BlobId"
},
"flowName": {
"type": [
"string",
"null"
]
},
"status": {
"type": "string"
},
"items": {
"$ref": "#/$defs/ItemStatistics"
},
"createdAt": {
"type": "string"
},
"completedAt": {
"type": [
"string",
"null"
]
},
"results": {
"type": [
"array",
"null"
],
"items": {
"$ref": "#/$defs/ItemResult"
}
}
},
"required": [
"runId",
"flowId",
"status",
"items",
"createdAt"
]
},
"ItemStatistics": {
"description": "Statistics about items in a run.",
"type": "object",
"properties": {
"total": {
"description": "Total number of items in this run.",
"type": "integer",
"format": "uint",
"minimum": 0
},
"completed": {
"description": "Number of completed items.",
"type": "integer",
"format": "uint",
"minimum": 0
},
"running": {
"description": "Number of currently running items.",
"type": "integer",
"format": "uint",
"minimum": 0
},
"failed": {
"description": "Number of failed items.",
"type": "integer",
"format": "uint",
"minimum": 0
},
"cancelled": {
"description": "Number of cancelled items.",
"type": "integer",
"format": "uint",
"minimum": 0
}
},
"required": [
"total",
"completed",
"running",
"failed",
"cancelled"
]
},
"ItemResult": {
"description": "Result for an individual item in a multi-item run.",
"type": "object",
"properties": {
"itemIndex": {
"description": "Index of this item in the input array (0-based).",
"type": "integer",
"format": "uint",
"minimum": 0
},
"status": {
"description": "Execution status of this item.",
"$ref": "#/$defs/ExecutionStatus"
},
"result": {
"description": "Result of this item, if completed.",
"oneOf": [
{
"$ref": "#/$defs/FlowResult"
},
{
"type": "null"
}
]
},
"completedAt": {
"description": "When this item completed (if completed).",
"type": [
"string",
"null"
],
"format": "date-time"
}
},
"required": [
"itemIndex",
"status"
]
},
"ExecutionStatus": {
"description": "Status of a workflow execution\n\nCases:\n* `running`: Execution is currently running\n* `completed`: Execution completed successfully\n* `failed`: Execution failed with an error\n* `cancelled`: Execution was cancelled by user request\n* `paused`: Execution is paused (debug mode)\n* `recoveryFailed`: Execution failed during recovery after orchestrator restart",
"type": "string",
"enum": [
"running",
"completed",
"failed",
"cancelled",
"paused",
"recoveryFailed"
]
},
"FlowResult": {
"oneOf": [
{
"$ref": "#/$defs/FlowResultSuccess"
},
{
"$ref": "#/$defs/FlowResultFailed"
}
],
"discriminator": {
"propertyName": "outcome",
"mapping": {
"success": "#/$defs/FlowResultSuccess",
"failed": "#/$defs/FlowResultFailed"
}
}
},
"FlowResultSuccess": {
"description": "The step execution was successful.",
"type": "object",
"properties": {
"outcome": {
"type": "string",
"const": "success",
"default": "success"
},
"result": {
"$ref": "#/$defs/Value"
}
},
"required": [
"outcome",
"result"
]
},
"FlowResultFailed": {
"description": "The step failed with the given error.",
"type": "object",
"properties": {
"outcome": {
"type": "string",
"const": "failed",
"default": "failed"
},
"error": {
"$ref": "#/$defs/FlowError"
}
},
"required": [
"outcome",
"error"
]
},
"FlowError": {
"description": "An error reported from within a flow or step.",
"type": "object",
"properties": {
"code": {
"type": "integer",
"format": "int64"
},
"message": {
"type": "string"
},
"data": {
"oneOf": [
{
"$ref": "#/$defs/Value"
},
{
"type": "null"
}
]
}
},
"required": [
"code",
"message"
]
},
"InitializeParams": {
"description": "Sent from Stepflow to the component server to begin the initialization process.",
"type": "object",
"properties": {
"runtimeProtocolVersion": {
"description": "Maximum version of the protocol being used by the Stepflow runtime.",
"type": "integer",
"format": "uint32",
"minimum": 0
},
"observability": {
"description": "Observability context for tracing initialization (trace context only, no flow/run).",
"oneOf": [
{
"$ref": "#/$defs/ObservabilityContext"
},
{
"type": "null"
}
]
},
"capabilities": {
"description": "Runtime capabilities provided to the component server.",
"oneOf": [
{
"$ref": "#/$defs/RuntimeCapabilities"
},
{
"type": "null"
}
]
}
},
"required": [
"runtimeProtocolVersion"
]
},
"RuntimeCapabilities": {
"description": "Runtime capabilities advertised by the Stepflow runtime during initialization.\n\nComponent servers can use these capabilities to access runtime services.",
"type": "object",
"properties": {
"blobApiUrl": {
"description": "Base URL for the Blob HTTP API.\n\nWhen provided, component servers should use direct HTTP requests\n(`GET {blob_api_url}/{blob_id}`, `POST {blob_api_url}`) for blob operations\ninstead of SSE bidirectional protocol.",
"type": [
"string",
"null"
]
},
"blobThreshold": {
"description": "Byte size threshold for automatic blobification.\n\nWhen set to a non-zero value, the orchestrator may replace large input fields\nwith `$blob` references. Component servers that report `supports_blob_refs`\nshould resolve these references before processing. Component servers should\nalso blobify output fields exceeding this threshold.\n\nA value of 0 or `None` means automatic blobification is disabled.",
"type": [
"integer",
"null"
],
"format": "int64"
}
}
},
"InitializeResult": {
"description": "Sent from the component server back to Stepflow with the result of initialization.\nThe component server will not be initialized until it receives the `initialized` notification.",
"type": "object",
"properties": {
"serverProtocolVersion": {
"description": "Version of the protocol being used by the component server.",
"type": "integer",
"format": "uint32",
"minimum": 0
},
"supportsBlobRefs": {
"description": "Whether this component server supports `$blob` references in inputs/outputs.\n\nWhen `true`, the orchestrator may send `$blob` references in component inputs\nand expects the server to resolve them. The server may also return `$blob`\nreferences in outputs for the orchestrator to resolve.\n\nWhen `false` (default), the orchestrator will not send blob refs and will\nresolve any refs before delivering input to this server.",
"type": "boolean"
}
},
"required": [
"serverProtocolVersion"
]
},
"Initialized": {
"description": "Sent from Stepflow to the component server after initialization is complete.",
"type": "object"
},
"ErrorCode": {
"description": "Well-known error codes for the Stepflow platform.\n\nRanges:\n -32700 to -32600: JSON-RPC standard errors\n -32000 to -32099: Worker errors (never retried)\n -32100 to -32199: Component execution errors (retried with onError: retry)\n -32200 to -32299: Orchestrator errors (never retried)\n -32300 to -32399: Transport errors (always retried)\n\nWorker predefined: -32000 to -32010, user-defined: -32011 to -32099.\nComponent execution predefined: -32100 to -32109, user-defined: -32110 to -32199.",
"type": "integer",
"enum": [
-32700,
-32600,
-32601,
-32602,
-32603,
-32000,
-32001,
-32002,
-32003,
-32004,
-32005,
-32006,
-32007,
-32008,
-32100,
-32101,
-32102,
-32103,
-32200,
-32201,
-32202,
-32300,
-32301,
-32302,
-32303
],
"x-enum-varnames": [
"PARSE_ERROR",
"JSON_RPC_INVALID_REQUEST",
"METHOD_NOT_FOUND",
"INVALID_PARAMS",
"JSON_RPC_INTERNAL_ERROR",
"WORKER_ERROR",
"COMPONENT_NOT_FOUND",
"WORKER_NOT_INITIALIZED",
"INVALID_INPUT_SCHEMA",
"INVALID_VALUE",
"NOT_FOUND",
"PROTOCOL_VERSION_MISMATCH",
"WORKER_DEPENDENCY_ERROR",
"WORKER_CONFIGURATION_ERROR",
"COMPONENT_EXECUTION_FAILED",
"COMPONENT_VALUE_ERROR",
"COMPONENT_RESOURCE_UNAVAILABLE",
"COMPONENT_BAD_REQUEST",
"UNDEFINED_FIELD",
"ENTITY_NOT_FOUND",
"INTERNAL_ERROR",
"TRANSPORT_ERROR",
"TRANSPORT_SPAWN_ERROR",
"TRANSPORT_CONNECTION_ERROR",
"TRANSPORT_PROTOCOL_ERROR"
]
},
"MethodRequest": {
"description": "Request to execute a method.",
"type": "object",
"properties": {
"jsonrpc": {
"$ref": "#/$defs/JsonRpc",
"default": "2.0"
},
"id": {
"$ref": "#/$defs/RequestId"
},
"method": {
"description": "The method being called.",
"$ref": "#/$defs/Method"
},
"params": {
"description": "The parameters for the method call. Set on method requests."
}
},
"required": [
"id",
"method"
]
},
"JsonRpc": {
"description": "The version of the JSON-RPC protocol.",
"type": "string",
"const": "2.0",
"default": "2.0"
},
"RequestId": {
"description": "The identifier for a JSON-RPC request. Can be either a string or an integer.\nThe RequestId is used to match method responses to corresponding requests.\nIt should not be set on notifications.",
"anyOf": [
{
"type": "string"
},
{
"type": "integer",
"format": "int64"
}
]
},
"Method": {
"type": "string",
"enum": [
"initialize",
"initialized",
"components/list",
"components/info",
"components/execute",
"components/infer_schema",
"runs/submit",
"runs/get"
]
},
"MethodResponse": {
"description": "Response to a method request. Success responses have a 'result' field while error responses have an 'error' field.",
"oneOf": [
{
"$ref": "#/$defs/MethodSuccess"
},
{
"$ref": "#/$defs/MethodError"
}
]
},
"MethodSuccess": {
"description": "The result of a successful method execution.",
"type": "object",
"properties": {
"jsonrpc": {
"$ref": "#/$defs/JsonRpc",
"default": "2.0"
},
"id": {
"$ref": "#/$defs/RequestId"
},
"result": {
"description": "The result of a successful method execution."
}
},
"required": [
"id",
"result"
]
},
"MethodError": {
"type": "object",
"properties": {
"jsonrpc": {
"$ref": "#/$defs/JsonRpc",
"default": "2.0"
},
"id": {
"$ref": "#/$defs/RequestId"
},
"error": {
"description": "An error that occurred during method execution.",
"$ref": "#/$defs/Error"
}
},
"required": [
"id",
"error"
]
},
"Error": {
"description": "An error returned from a method execution.",
"type": "object",
"properties": {
"code": {
"description": "A numeric code indicating the error type.",
"type": "integer",
"format": "int64"
},
"message": {
"description": "Concise, single-sentence description of the error.",
"type": "string"
},
"data": {
"description": "Primitive or structured value that contains additional information about the error."
}
},
"required": [
"code",
"message"
]
},
"Notification": {
"description": "A JSON-RPC notification message.",
"type": "object",
"properties": {
"jsonrpc": {
"$ref": "#/$defs/JsonRpc"
},
"method": {
"description": "The notification method being called.",
"type": "string"
},
"params": {
"description": "The parameters for the notification."
}
},
"required": [
"method"
]
}
}
}