Building a Python External Task Worker for Camunda

The External Task Pattern: Decoupling Process and Implementation

In enterprise architecture, it's crucial to separate the what from the how. The BPMN model defines the business process logic (what needs to happen), while external workers implement the specific technical tasks (how they are done). This decoupling is powerful; you can update the implementation of a service (e.g., switch from a legacy SOAP API to a modern gRPC service) without ever touching the business process model.

The interaction works as follows:

1. A process instance in the Camunda engine reaches a "Service Task" configured as an External Task.
2. The engine doesn't execute anything. It simply creates a work item on a specific Topic and waits.
3. An External Worker—a separate application, which we will write in Python—is subscribed to that topic. It periodically polls the engine via its REST API, asking, "Are there any jobs for me?" This is called a "fetch and lock."
4. The engine assigns a task to the worker, locking it to prevent other workers from processing the same item.
5. The worker executes its business logic (e.g., calling an API, running a script, performing a calculation).
6. Upon completion, the worker notifies the engine, sending back any resulting data. The engine then moves the process forward.

The following diagram perfectly illustrates this client-server interaction between the worker and the engine.

This diagram shows the communication flow for the External Task pattern. Our Python script acts as "Some Worker," which polls the Camunda engine for tasks on a given topic using a fetchAndLock call and reports results using a complete call.

Setting Up Your Environment

To implement this, you need two main components: a running Camunda engine and a Python library to communicate with it.

1. Running the Camunda Engine

The easiest way to run a local Camunda instance for development is using Docker. The camunda-external-task-client-python3 project, which we will use, provides a convenient docker-compose file for this purpose.

2. Installing the Python Client

Next, you need to install the Python library that will handle the REST API communication for you.

Hands-On: Building the External Task Worker

Now, let's write the Python code for our worker. This worker will handle the "Check-in Truck" task from our BPMN diagram. Its job will be to simulate a call to the Navis N4 system to validate a truck's appointment.

1. Mocking the Navis N4 API

In a real system, your worker would make an HTTP request to the Navis N4 API. For development and testing, it's standard practice to use a mock service. A mock simulates the behavior of the real API without requiring a connection to the actual system. This isolates your component for testing and avoids dependencies on external systems that may be unavailable.

This diagram shows the difference between testing against a mock server and developing against a real API server. Our approach in this lesson mirrors the "Testing" scenario, where our worker acts as the client and communicates with a self-contained mock function.

Create a new Python file, e.g., gate_worker.py. Inside it, let's create a simple function to simulate this API call.

import random
import time

def mock_navis_n4_check_in(truck_id: str, appointment_id: str) -> dict:
    """
    Simulates an API call to Navis N4 to check in a truck.
    In a real scenario, this would be an HTTP request.
    """
    print(f"  [Mock Navis N4] Checking in truck {truck_id} with appointment {appointment_id}...")
    
    # Simulate network latency and processing time
    time.sleep(random.uniform(0.5, 2.0))

    # Simulate potential API failures or business outcomes
    if random.random() < 0.1: # 10% chance of failure
        print(f"  [Mock Navis N4] Check-in failed for truck {truck_id}.")
        return {"status": "FAILED", "reason": "Invalid appointment ID"}
    
    print(f"  [Mock Navis N4] Check-in successful for truck {truck_id}.")
    return {"status": "SUCCESS", "gate_pass_id": f"GP-{random.randint(10000, 99999)}"}

This function mimics key aspects of a real API call: it takes inputs, simulates delay, and can return different success or failure responses.

2. Implementing the Worker Logic

Now, let's build the Camunda worker that uses this mock function. We'll follow the structure provided in the client library's documentation.

Let's integrate these concepts into our gate_worker.py script. Add the following code below your mock function:

from camunda.external_task.external_task import ExternalTask, TaskResult
from camunda.external_task.external_task_worker import ExternalTaskWorker

# Configuration for the worker
default_config = {
    "maxTasks": 1,
    "lockDuration": 10000,
    "asyncResponseTimeout": 5000,
    "retries": 3,
    "retryTimeout": 5000,
    "sleepSeconds": 10  # Poll every 10 seconds
}

def handle_truck_check_in(task: ExternalTask) -> TaskResult:
    """
    Handler for the 'check-in-truck' topic.
    """
    print(f"Received task {task.get_task_id()} for topic 'check-in-truck'")

    # Get process variables sent from Camunda
    truck_id = task.get_variable("truckId")
    appointment_id = task.get_variable("appointmentId")
    
    if not truck_id or not appointment_id:
        print("Task failed: Missing required variables 'truckId' or 'appointmentId'")
        return task.failure(error_message="Missing required variables", 
                            error_details="The process must provide truckId and appointmentId.")

    # Call our business logic (the mock API)
    result = mock_navis_n4_check_in(truck_id, appointment_id)

    # Prepare variables to send back to the process
    output_vars = {
        "checkInStatus": result.get("status"),
        "gatePassId": result.get("gate_pass_id"),
        "checkInFailureReason": result.get("reason")
    }
    
    # Complete the task and send back the results
    print(f"Completing task {task.get_task_id()} with status: {result.get('status')}")
    return task.complete(output_vars)

if __name__ == '__main__':
    print("Starting Gate Worker...")
    # The worker subscribes to the 'check-in-truck' topic and assigns the handler
    ExternalTaskWorker(worker_id="gate-worker-1", config=default_config).subscribe("check-in-truck", handle_truck_check_in)

This script now defines a complete worker. It subscribes to the check-in-truck topic, calls our mock function with variables from the process, and returns the results. Your experience with designing system interactions will highlight the clean separation of concerns here: the worker is a self-contained component focused entirely on one business capability.

Running the End-to-End Workflow

Let's see it all in action.

1. Start Camunda: Ensure your Camunda Docker container is running (docker-compose up).
2. Update BPMN Model:
   • Open your gate-in-process.bpmn file in the Camunda Modeler.
   • Select the task that represents the truck check-in (e.g., "Check-in Truck at Gate").
   • In the Properties Panel on the right, change the Implementation dropdown to External.
   • Set the Topic field to check-in-truck. This string must exactly match the topic in your Python script's subscribe call.
   • Save the diagram.
3. Deploy the Process:
   • Click the "Deploy" button in the Camunda Modeler (looks like a paper airplane).
   • Give it a deployment name (e.g., "Gate In Process with Worker") and ensure the target engine URL is correct (http://localhost:8080/engine-rest). Click "Deploy".
4. Run the Python Worker:
   • In your terminal, run the Python script: python gate_worker.py.
   • You should see the message "Starting Gate Worker...". It is now polling the engine for jobs.
5. Start a Process Instance:
   • Open the Camunda Tasklist: http://localhost:8080/camunda/app/tasklist/.
   • Click "Start process" and select your newly deployed process.
   • You will be asked to provide initial variables. Create two variables:
     • truckId (String): e.g., T-12345
     • appointmentId (String): e.g., APT-ABC-987
   • Click "Start".
6. Observe the Magic:
   • In your Python terminal: You will see the worker spring to life. It will print messages indicating it received a task, called the mock API, and completed the task.
   • In Camunda Cockpit: Navigate to Cockpit (http://localhost:8080/camunda/app/cockpit/) and find your running process instance. You'll see that the token has moved past the "Check-in Truck" task. If you inspect the process variables for that instance, you will now see the new variables (checkInStatus, gatePassId) that were returned by your Python worker.

For further insight into building workers, the article "Implementing an external task worker for Camunda in Python" by Peter Klauke provides an excellent walkthrough of creating a worker from scratch using a slightly lower-level library, PyCamunda. It reinforces the core concepts of fetching, locking, and completing tasks.

Conclusion

Congratulations! You have successfully implemented a critical architectural pattern for workflow automation. You have decoupled the business process from the technical implementation by creating a Python external task worker that subscribes to topics from the Camunda engine. This worker executes business logic—simulated via a mock API call—and communicates the results back to the process, influencing its path forward. This component-based, message-driven architecture is fundamental to building scalable and maintainable enterprise systems.

In the next lesson, we will complete the integration loop. You will connect the SimPy simulation from the previous lesson to this Camunda workflow. Instead of manually starting process instances, the events from your simulation (i.e., a truck arriving) will programmatically trigger new workflow instances via Camunda's REST API, creating a truly dynamic, end-to-end process simulation.