Temporal.io Python SDK

import asyncio
from datetime import timedelta
from temporalio.client import Client
from temporalio.worker import Worker
from temporalio import activity, workflow

# Define an activity
@activity.defn
async def say_hello(name: str) -> str:
    return f"Hello, {name}!"

# Define a workflow
@workflow.defn
class GreetingWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            say_hello,
            name,
            schedule_to_close_timeout=timedelta(seconds=5),
        )

async def main():
    # Connect to Temporal server (default to localhost:7233)
    # Use os.environ.get('TEMPORAL_HOST_PORT', 'localhost:7233') for dynamic connection
    client = await Client.connect(os.environ.get('TEMPORAL_HOST_PORT', 'localhost:7233'))

    # Run a worker
    task_queue_name = "my-task-queue"
    worker = Worker(
        client,
        task_queue=task_queue_name,
        workflows=[GreetingWorkflow],
        activities=[say_hello],
    )
    # Start worker in background
    worker_task = asyncio.create_task(worker.run())
    print(f"Worker started on task queue '{task_queue_name}'...")

    # Start a workflow execution
    result = await client.execute_workflow(
        GreetingWorkflow.run,
        "Temporal",
        id="greeting-workflow-id",
        task_queue=task_queue_name,
    )
    print(f"Workflow result: {result}") # Expected: "Hello, Temporal!"

    # Clean up worker
    worker_task.cancel()
    await worker_task

if __name__ == "__main__":
    import os
    asyncio.run(main())