# Stepflow Python SDK
Python SDK for building Stepflow components and workflows.
## Installation
```bash
# Install from source
uv add stepflow-py
```
## Usage
### Creating a Component Server
```python
from stepflow_py import StepflowStdioServer, StepflowContext
import msgspec
# Define input/output types
class MyInput(msgspec.Struct):
message: str
count: int
class MyOutput(msgspec.Struct):
result: str
# Create server
server = StepflowStdioServer()
# Register a component
@server.component
def my_component(input: MyInput) -> MyOutput:
return MyOutput(result=f"Processed: {input.message} x{input.count}")
# Component with context (for blob operations)
@server.component
async def component_with_context(input: MyInput, context: StepflowContext) -> MyOutput:
# Store data as a blob
blob_id = await context.put_blob({"processed": input.message})
return MyOutput(result=f"Stored blob: {blob_id}")
# Run the server
if __name__ == "__main__":
server.run()
```
### Using the Context API
The `StepflowContext` provides bidirectional communication with the Stepflow runtime:
```python
# Store JSON data as a blob
blob_id = await context.put_blob({"key": "value"})
# Retrieve blob data
data = await context.get_blob(blob_id)
# Logging
context.log("Debug message")
```
## Development
### Running Tests
```bash
uv run pytest
```
### Type Checking
```bash
uv run mypy src/
```
### Protocol Generation
This SDK uses auto-generated protocol types from the JSON schema. To regenerate the protocol types when the schema changes:
```bash
uv run python generate.py
```
The generation script automatically handles the generation and applies necessary fixes for msgspec compatibility.
### Project Structure
- `src/stepflow_py/` - Main SDK code
- `generated_protocol.py` - Auto-generated protocol types from JSON schema
- `protocol.py` - Hybrid protocol layer with envelope patterns for efficient deserialization
- `server.py` - Component server implementation
- `context.py` - Runtime context API
- `exceptions.py` - SDK-specific exceptions
- `tests/` - Test suite
### Architecture
The SDK uses a hybrid approach for protocol handling:
1. **Generated Types** (`generated_protocol.py`) - Auto-generated from the Stepflow JSON schema using `datamodel-code-generator`
2. **Protocol Layer** (`protocol.py`) - Combines generated types with manual envelope patterns for two-stage deserialization using `msgspec.Raw`
3. **Server Implementation** - Uses the protocol layer for efficient JSON-RPC message handling
This design provides:
- **Type Safety** - All protocol messages use properly typed structs
- **Schema Consistency** - Generated types match the Rust protocol exactly
- **Performance** - Two-stage deserialization with `msgspec.Raw` for optimal speed
- **Maintainability** - Protocol changes can be regenerated automatically
## License
Licensed under the Apache License, Version 2.0.
Raw data
{
"_id": null,
"home_page": null,
"name": "stepflow-py",
"maintainer": "DataStax Inc.",
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": "ai, automation, components, orchestration, pipeline, stepflow, workflow",
"author": "DataStax Inc.",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/30/e1/9bb4e566840c09efc7bbb159ae1b27af2632bb1c4b7dcaaaa7f3b4cf2adb/stepflow_py-0.6.0.tar.gz",
"platform": null,
"description": "# Stepflow Python SDK\n\nPython SDK for building Stepflow components and workflows.\n\n## Installation\n\n```bash\n# Install from source\nuv add stepflow-py\n```\n\n## Usage\n\n### Creating a Component Server\n\n```python\nfrom stepflow_py import StepflowStdioServer, StepflowContext\nimport msgspec\n\n# Define input/output types\nclass MyInput(msgspec.Struct):\n message: str\n count: int\n\nclass MyOutput(msgspec.Struct):\n result: str\n\n# Create server\nserver = StepflowStdioServer()\n\n# Register a component\n@server.component\ndef my_component(input: MyInput) -> MyOutput:\n return MyOutput(result=f\"Processed: {input.message} x{input.count}\")\n\n# Component with context (for blob operations)\n@server.component\nasync def component_with_context(input: MyInput, context: StepflowContext) -> MyOutput:\n # Store data as a blob\n blob_id = await context.put_blob({\"processed\": input.message})\n return MyOutput(result=f\"Stored blob: {blob_id}\")\n\n# Run the server\nif __name__ == \"__main__\":\n server.run()\n```\n\n### Using the Context API\n\nThe `StepflowContext` provides bidirectional communication with the Stepflow runtime:\n\n```python\n# Store JSON data as a blob\nblob_id = await context.put_blob({\"key\": \"value\"})\n\n# Retrieve blob data\ndata = await context.get_blob(blob_id)\n\n# Logging\ncontext.log(\"Debug message\")\n```\n\n## Development\n\n### Running Tests\n\n```bash\nuv run pytest\n```\n\n### Type Checking\n\n```bash\nuv run mypy src/\n```\n\n### Protocol Generation\n\nThis SDK uses auto-generated protocol types from the JSON schema. To regenerate the protocol types when the schema changes:\n\n```bash\nuv run python generate.py\n```\n\nThe generation script automatically handles the generation and applies necessary fixes for msgspec compatibility.\n\n### Project Structure\n\n- `src/stepflow_py/` - Main SDK code\n - `generated_protocol.py` - Auto-generated protocol types from JSON schema\n - `protocol.py` - Hybrid protocol layer with envelope patterns for efficient deserialization\n - `server.py` - Component server implementation\n - `context.py` - Runtime context API\n - `exceptions.py` - SDK-specific exceptions\n- `tests/` - Test suite\n\n### Architecture\n\nThe SDK uses a hybrid approach for protocol handling:\n\n1. **Generated Types** (`generated_protocol.py`) - Auto-generated from the Stepflow JSON schema using `datamodel-code-generator`\n2. **Protocol Layer** (`protocol.py`) - Combines generated types with manual envelope patterns for two-stage deserialization using `msgspec.Raw`\n3. **Server Implementation** - Uses the protocol layer for efficient JSON-RPC message handling\n\nThis design provides:\n- **Type Safety** - All protocol messages use properly typed structs\n- **Schema Consistency** - Generated types match the Rust protocol exactly\n- **Performance** - Two-stage deserialization with `msgspec.Raw` for optimal speed\n- **Maintainability** - Protocol changes can be regenerated automatically\n\n## License\n\nLicensed under the Apache License, Version 2.0.",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "Python SDK for Stepflow components and workflows.",
"version": "0.6.0",
"project_urls": {
"Bug Tracker": "https://github.com/stepflow-ai/stepflow/issues",
"Changelog": "https://github.com/stepflow-ai/stepflow/blob/main/sdks/python/CHANGELOG.md",
"Documentation": "https://stepflow-ai.github.io/stepflow/",
"Homepage": "https://stepflow-ai.github.io/stepflow/",
"Repository": "https://github.com/stepflow-ai/stepflow",
"Source Code": "https://github.com/stepflow-ai/stepflow/tree/main/sdks/python"
},
"split_keywords": [
"ai",
" automation",
" components",
" orchestration",
" pipeline",
" stepflow",
" workflow"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "1f54a9314bf6189f0d5c57e842727f880d4a5e96fe4515dedbf7393694558eff",
"md5": "c7a49bc5ae04647f65909c1d71811c83",
"sha256": "feaab87a639b70725ad48229d009055dc1cf1a1031f435654472404eeda650ba"
},
"downloads": -1,
"filename": "stepflow_py-0.6.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "c7a49bc5ae04647f65909c1d71811c83",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 50438,
"upload_time": "2025-10-15T20:28:41",
"upload_time_iso_8601": "2025-10-15T20:28:41.390399Z",
"url": "https://files.pythonhosted.org/packages/1f/54/a9314bf6189f0d5c57e842727f880d4a5e96fe4515dedbf7393694558eff/stepflow_py-0.6.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "30e19bb4e566840c09efc7bbb159ae1b27af2632bb1c4b7dcaaaa7f3b4cf2adb",
"md5": "3b320316593283abe7ee002bab1ccd11",
"sha256": "2ba45af989723a0f1850016eca19fdd7c18096b3580e46294771c5d816d4f6be"
},
"downloads": -1,
"filename": "stepflow_py-0.6.0.tar.gz",
"has_sig": false,
"md5_digest": "3b320316593283abe7ee002bab1ccd11",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 137202,
"upload_time": "2025-10-15T20:28:42",
"upload_time_iso_8601": "2025-10-15T20:28:42.484445Z",
"url": "https://files.pythonhosted.org/packages/30/e1/9bb4e566840c09efc7bbb159ae1b27af2632bb1c4b7dcaaaa7f3b4cf2adb/stepflow_py-0.6.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-15 20:28:42",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "stepflow-ai",
"github_project": "stepflow",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "stepflow-py"
}
JSON
