The lightweight observability layer for Python scripts and pipelines. Stream logs, monitor progress, and visualize execution in real-time.
Every Python script deserves a voice.
stepcast wraps any Python function in a named, observable step. When a pipeline runs, each step announces itself β label, live output, duration, pass/fail β in the terminal, Colab notebook, or your local web dashboard.
Zero required dependencies. Works on every OS. Readable by everyone.
| Feature | Details |
|---|---|
| π‘ Live output streaming | Every print() inside a step streams to terminal in real time |
| β Pass / β Fail / β Skip | Clear visual status with timing for every step |
| π Retry with backoff | retries=3, retry_delay=2.0 with exponential backoff |
| β±οΈ Timeout enforcement | Per-step timeout raises StepFailedError cleanly |
| π¨ Dry run mode | Preview all steps without executing anything |
| π€ Gemini AI narration | Optional: explains what each step did in plain English |
| π Local web dashboard | stepcast serve β run history at localhost:4321 |
| π³ Docker self-hosted | One command team dashboard (docker-compose up) |
| π Google Colab | Native auth + save to Drive |
| π¨ Rich support | Beautiful spinners + colour when rich is installed |
| π i18n ready | All strings in locale files, community-translated |
| π Zero dep core | Pure stdlib β pip install stepcast just works |
# Core library (zero dependencies)
pip install stepcast
# Add beautiful terminal formatting
pip install "stepcast[rich]"
# Add the local web dashboard
pip install "stepcast[dashboard]"
# Add AI narration using Google Gemini
pip install "stepcast[gemini]"
# Install everything!
pip install "stepcast[all]"Building an observable pipeline takes seconds:
from stepcast import Pipeline
pipe = Pipeline("My First Pipeline")
@pipe.step("Download data", retries=2)
def download():
print("Fetching records...")
return [1, 2, 3]
@pipe.step("Process data")
def process(data: list):
result = sum(data)
print(f"Sum = {result}")
return result
@pipe.step("Save results", skip_if=lambda: False)
def save(total: int):
print(f"Saving total: {total}")
report = pipe.run()
print(report.summary())What you see in the terminal:
βββββββββββββββββββββββββββββββββββββββββ
π‘ My First Pipeline
βββββββββββββββββββββββββββββββββββββββββ
βΆ Download data...
β Fetching records...
β
Done (0.01s)
βΆ Process data...
β Sum = 6
β
Done (0.00s)
βΆ Save results...
β Saving total: 6
β
Done (0.00s)
βββββββββββββββββββββββββββββββββββββββββ
β
3 of 3 steps passed Total: 0.01s
βββββββββββββββββββββββββββββββββββββββββ
Monitor your runs graphically with the built-in local dashboard!
stepcast serve
# β Opens http://localhost:4321 automatically
# β Shows your run history, step details, analyticsStream a live run:
pipe = Pipeline("ETL Job", dashboard=True)
pipe.run() # Watch it run live in your browser!stepcast can use Google Gemini 2.5 Flash to automatically summarize what your code did in plain English. Ideal for long builds or sharing logs with non-technical team members!
# Get your free key from Google AI Studio
export STEPCAST_GEMINI_API_KEY="AIza..."pipe = Pipeline("Data Cruncher", narrate=True)
# Terminal Output:
# βΆ Clean Dataset...
# π¬ "The pipeline removed 45 invalid rows and imputed missing values in 1.2s."
# β
Donestepcast version # Print installed version
stepcast doctor # Diagnose your environment (Python, OS, packages, keys)
stepcast config set gemini_api_key YOUR_KEY
stepcast run script.py # Run a pipeline and ensure proper exit codesDive deeper into how stepcast works:
- π Quickstart & Core Concepts
- π Full API Reference
- π Dashboard Guide
- π€ AI Narration Setup
- π Colab Integration
- π‘οΈ Security Policy
- π Changelog
Contributions are highly welcome! See CONTRIBUTING.md for branch naming conventions, PR processes, and testing standards. We especially welcome translations for the locale files!
- οΏ½ Report a Bug
Released under the MIT License Β© 2026.
Built with passion for the global Python community by Suneel Bose K.