# 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": "StepFlow Contributors",
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": "ai, automation, components, orchestration, pipeline, stepflow, workflow",
"author": "StepFlow Contributors",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/c4/d2/12c060b85805764036f939f140d998d10f225cd5707c6c049d6b361bf79a/stepflow_py-0.2.4.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.2.4",
"project_urls": {
"Bug Tracker": "https://github.com/riptano/stepflow/issues",
"Changelog": "https://github.com/riptano/stepflow/blob/main/sdks/python/CHANGELOG.md",
"Documentation": "https://riptano.github.io/stepflow/",
"Homepage": "https://riptano.github.io/stepflow/",
"Repository": "https://github.com/riptano/stepflow",
"Source Code": "https://github.com/riptano/stepflow/tree/main/sdks/python"
},
"split_keywords": [
"ai",
" automation",
" components",
" orchestration",
" pipeline",
" stepflow",
" workflow"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "7c3d68f589aecaef8045660fad1d1571de005d2a62ca3d61996f206947f133a2",
"md5": "d06cec72c6044a4ce7a7770d628b756e",
"sha256": "3b24ddeace9700b2dd29368601366ffc49098aa47045e9e86729cbb39c94c7f6"
},
"downloads": -1,
"filename": "stepflow_py-0.2.4-py3-none-any.whl",
"has_sig": false,
"md5_digest": "d06cec72c6044a4ce7a7770d628b756e",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 35255,
"upload_time": "2025-07-30T04:52:58",
"upload_time_iso_8601": "2025-07-30T04:52:58.859387Z",
"url": "https://files.pythonhosted.org/packages/7c/3d/68f589aecaef8045660fad1d1571de005d2a62ca3d61996f206947f133a2/stepflow_py-0.2.4-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "c4d212c060b85805764036f939f140d998d10f225cd5707c6c049d6b361bf79a",
"md5": "15a74c66f2b93215bccf090c57c2ea1b",
"sha256": "73ea6912f95153f32a03c43ddc6b46c1cbef79eb3bbb787bef4620eb9d95866b"
},
"downloads": -1,
"filename": "stepflow_py-0.2.4.tar.gz",
"has_sig": false,
"md5_digest": "15a74c66f2b93215bccf090c57c2ea1b",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 100531,
"upload_time": "2025-07-30T04:53:00",
"upload_time_iso_8601": "2025-07-30T04:53:00.234496Z",
"url": "https://files.pythonhosted.org/packages/c4/d2/12c060b85805764036f939f140d998d10f225cd5707c6c049d6b361bf79a/stepflow_py-0.2.4.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-30 04:53:00",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "riptano",
"github_project": "stepflow",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "stepflow-py"
}
JSON
