Skip to main content

Unofficial Python SDK for the Temporal Workflow Engine

Project description

Unofficial Python SDK for the Temporal Workflow Engine


This should be considered EXPERIMENTAL at the moment. At the moment, all I can say is that the test cases currently pass. I have not tested this for any real world use cases yet.


pip install temporal-python-sdk

Sample Code

Sample code for using this library can be found in Workflows in Python Using Temporal.

Hello World

import asyncio
import logging
from datetime import timedelta

from temporal.activity_method import activity_method
from temporal.workerfactory import WorkerFactory
from temporal.workflow import workflow_method, Workflow, WorkflowClient


TASK_QUEUE = "HelloActivity-python-tq"
NAMESPACE = "default"

# Activities Interface
class GreetingActivities:
    @activity_method(task_queue=TASK_QUEUE, schedule_to_close_timeout=timedelta(seconds=1000))
    async def compose_greeting(self, greeting: str, name: str) -> str:
        raise NotImplementedError

# Activities Implementation
class GreetingActivitiesImpl:
    async def compose_greeting(self, greeting: str, name: str):
        return greeting + " " + name + "!"

# Workflow Interface
class GreetingWorkflow:
    async def get_greeting(self, name: str) -> str:
        raise NotImplementedError

# Workflow Implementation
class GreetingWorkflowImpl(GreetingWorkflow):

    def __init__(self):
        self.greeting_activities: GreetingActivities = Workflow.new_activity_stub(GreetingActivities)

    async def get_greeting(self, name):
        return await self.greeting_activities.compose_greeting("Hello", name)

async def client_main():
    client = WorkflowClient.new_client(namespace=NAMESPACE)

    factory = WorkerFactory(client, NAMESPACE)
    worker = factory.new_worker(TASK_QUEUE)
    worker.register_activities_implementation(GreetingActivitiesImpl(), "GreetingActivities")

    greeting_workflow: GreetingWorkflow = client.new_workflow_stub(GreetingWorkflow)
    result = await greeting_workflow.get_greeting("Python")
    print("Stopping workers.....")
    await worker.stop()
    print("Workers stopped......")

if __name__ == '__main__':



  • Workflow argument passing and return values
  • Activity invocation
  • Activity heartbeat and Activity.getHeartbeatDetails()
  • doNotCompleteOnReturn
  • ActivityCompletionClient
    • complete
    • complete_exceptionally
  • Activity get_namespace(), get_task_token() get_workflow_execution()
  • Activity Retry
  • Activity Failure Exceptions
  • workflow_execution_timeout / workflow_run_timeout / workflow_task_timeout
  • Workflow exceptions
  • Cron workflows
  • Workflow static methods:
    • await_till()
    • sleep()
    • current_time_millis()
    • now()
    • random_uuid()
    • new_random()
    • get_workflow_id()
    • get_run_id()
    • get_version()
    • get_logger()
  • Activity invocation parameters
  • Query method
  • Signal methods
  • Workflow start parameters - workflow_id etc...
  • Workflow client - starting workflows synchronously
  • Workflow client - starting workflows asynchronously (WorkflowClient.start)
  • Get workflow result after async execution (client.wait_for_close)
  • Workflow client - invoking signals
  • Workflow client - invoking queries


  • ActivityStub and Workflow.newUntypedActivityStub
  • Remove threading, use coroutines for everything all concurrency
  • Classes as arguments and return values to/from activity and workflow methods (DataConverter)
    • Type hints for DataConverter
  • Parallel activity execution (STATUS: there's a working but not finalized API).


  • Timers
  • Custom workflow ids through start() and new_workflow_stub()
  • WorkflowStub and WorkflowClient.newUntypedWorkflowStub
  • ContinueAsNew status-wip
  • Sticky workflows status-wip
  • Child Workflows status-wip


  • Support for keyword arguments


  • Compatibility with Java client
  • Compatibility with Golang client
  • Upgrade python-betterproto

Post 2.0:

  • sideEffect/mutableSideEffect
  • Local activity
  • Cancellation Scopes
  • Explicit activity ids for activity invocations

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

temporal-python-sdk-1.0.19.tar.gz (51.6 kB view hashes)

Uploaded Source

Built Distribution

temporal_python_sdk-1.0.19-py3-none-any.whl (64.4 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page