Temporal-boost is a lightweight, high-level framework for rapid development of Temporal-based microservices in Python. Built on top of the official Temporal Python SDK, it provides a FastAPI-inspired developer experience.
📖 Full Documentation | 🐛 Issues | 💬 Discussions
- ✅ FastAPI-style API - Organize workers like FastAPI routes
- ✅ Zero boilerplate - Focus on business logic, not infrastructure
- ✅ CRON workers - Scheduled workflows with one line of code
- ✅ ASGI integration - Run FastAPI alongside Temporal workers
- ✅ FastStream support - Event-driven architectures
- ✅ Production-ready - Built-in logging, metrics, and graceful shutdown
- ✅ Type-safe - Full type hints and Pydantic integration
- Python >= 3.10
- Access to a Temporal server (local or remote)
pip install temporal-boost
# or
poetry add temporal-boost# FastStream integration
pip install "temporal-boost[faststream]"
# ASGI server support (choose one or more)
pip install "temporal-boost[uvicorn]"
pip install "temporal-boost[hypercorn]"
pip install "temporal-boost[granian]"import logging
from datetime import timedelta
from temporalio import activity, workflow
from temporal_boost import BoostApp
logging.basicConfig(level=logging.INFO)
app = BoostApp(name="my-service")
@activity.defn(name="greet_activity")
async def greet_activity(name: str) -> str:
return f"Hello, {name}!"
@workflow.defn(sandboxed=False, name="GreetingWorkflow")
class GreetingWorkflow:
@workflow.run
async def run(self, name: str) -> str:
return await workflow.execute_activity(
greet_activity,
name,
task_queue="greeting_queue",
start_to_close_timeout=timedelta(minutes=1),
)
app.add_worker(
"greeting_worker",
"greeting_queue",
activities=[greet_activity],
workflows=[GreetingWorkflow],
)
if __name__ == "__main__":
app.run()Run your application:
# Start all workers
python3 main.py run all
# Or run a specific worker
python3 main.py run greeting_workerAll configuration is handled via environment variables. See the Configuration Guide for complete details.
Common settings:
export TEMPORAL_TARGET_HOST=localhost:7233
export TEMPORAL_NAMESPACE=default
export TEMPORAL_USE_PYDANTIC_DATA_CONVERTER=trueWorker tuning:
export TEMPORAL_MAX_CONCURRENT_ACTIVITIES=300
export TEMPORAL_MAX_CONCURRENT_WORKFLOW_TASKS=300
export TEMPORAL_PROMETHEUS_BIND_ADDRESS=0.0.0.0:9090- 📖 Getting Started - Overview and installation
- 🏗️ Creating Applications - Activities, workflows, and workers
- 🚀 Running Applications - Deployment and production
- 🔧 Configuration - Complete configuration reference
- 💡 Examples - Comprehensive examples and patterns
- 🎯 Advanced Usage - Customization and advanced features
- 📚 API Reference - Complete API documentation
- 🔍 Troubleshooting - Common issues and solutions
# CRON worker
app.add_worker(
"daily_report",
"report_queue",
workflows=[DailyReportWorkflow],
cron_schedule="0 0 * * *",
cron_runner=DailyReportWorkflow.run,
)
# ASGI worker (FastAPI)
from fastapi import FastAPI
fastapi_app = FastAPI()
app.add_asgi_worker("api_worker", fastapi_app, "0.0.0.0", 8000)
# FastStream worker
from faststream import FastStream
faststream_app = FastStream(broker)
app.add_faststream_worker("message_worker", faststream_app)See Examples for more patterns.
MIT License - see LICENSE file for details.
Contributions are welcome! Please feel free to submit a Pull Request.