Register a Python Function, Run It From a Workflow, and Rerun Failed Runs

Picture the last place you ordered food at a counter. The cashier hands you a buzzer, and now you've got a choice. You can stand there staring at the kitchen like it owes you money, or you can take the buzzer to a table and scroll your phone until it goes off.

Either way the order slip does the same thing. It says what you want, and the right cook in the back picks it up because that's the dish they make. Nobody walks into the kitchen hunting for a person. The name on the slip does the routing.

That's how running a workflow from Python works with the Unmeshed SDK. Your function is the cook and the step name is the dish on the slip. You get the same two choices as the buzzer, wait for the result or come back later.

Install the SDK and connect

Install the package from PyPI.

pip install unmeshed-sdk

Then build a client. You need an API client ID and auth token from your Unmeshed instance, plus the URL and port where it runs. This example points at a local instance.

from unmeshed.sdk.configs.client_config import ClientConfig
from unmeshed.sdk.unmeshed_client import UnmeshedClient

client_config = ClientConfig()
client_config.set_client_id("your-client-id")
client_config.set_auth_token("your-auth-token")
client_config.set_base_url("http://localhost")
client_config.set_port(8080)

client_config.set_initial_delay_millis(50)
client_config.set_step_timeout_millis(3600000)
client_config.set_work_request_batch_size(200)
client_config.set_response_submit_batch_size(1000)
client_config.set_max_threads_count(10)
client_config.set_poll_interval_millis(10)

client = UnmeshedClient(client_config)

The first four settings say who you are and where to connect. The rest tune how the worker polls for work and how many threads it uses. Two of them matter later in this post.

Write the function and register it

A worker is a plain Python function. It takes a dict and returns a dict. Sync and async functions both work.

from unmeshed.sdk.apis.workers.worker import Worker

def sample_function(input_dict: dict) -> dict:
    return {
        "message": "Hello, world! sample_function",
        "input_received": input_dict,
    }

client.register_worker(
    Worker(
        execution_method=sample_function,
        name="sample_function",
        namespace="default",
        max_in_progress=100,
    )
)

client.start()

The name is the slip. When a process reaches a step called sample_function, the SDK calls your function with that step's input and sends the result back. Registering says which cook works which station, and client.start() opens the kitchen. From then on the worker listens for tasks.

If registration fails with a 403, check the client's Allowed Workers Policy in the platform. It's a permission on the credentials and needs no code change.

There's also a decorator version, @worker_function, that takes the same name, namespace and max_in_progress. Use whichever reads better in your codebase.

Add a step with the same name

The function won't run until a process has a step to call it. Here's a minimal process definition with one WORKER step. Import it, or build the same thing in the visual editor.

{
  "namespace": "default",
  "name": "test_process",
  "version": 1,
  "type": "API_ORCHESTRATION",
  "steps": [
    {
      "name": "sample_function",
      "type": "WORKER",
      "ref": "say_hello",
      "input": {
        "test1": "{{ context.input.test1 }}"
      }
    }
  ]
}

The step's name has to match the name you registered. That's the whole link between the two. The ref is the step's own ID inside the process, and the input line passes the request's test1 value into your function.

Run it and choose how to wait

Keep your worker running. In a second script, build the client the same way as in step one and run the process.

from unmeshed.sdk.common.process_request_data import ProcessRequestData

# Stand at the counter and wait.
sync_request = ProcessRequestData(
    namespace="default",
    name="test_process",
    version=1,
    requestId="req001",
    correlationId="corr001",
    input={"test1": "value", "test2": 100, "test3": 100.0},
)
process_data = client.run_process_sync(sync_request)
print(process_data.to_json())

# Take the buzzer and carry on.
async_request = ProcessRequestData(
    namespace="default",
    name="test_process",
    version=1,
    requestId="req002",
    correlationId="corr001",
    input={"test1": "value", "test2": 100, "test3": 100.0},
)
async_data = client.run_process_async(async_request)
print(async_data.to_json())

run_process_sync waits for the process to finish and hands back the result. run_process_async starts it and returns right away, which suits long jobs. The request ID identifies this request and the correlation ID ties related runs together, so give each run its own request ID.

When an order goes wrong

Every run gets a process ID, and that's your handle for everything that follows. Fetch a run to see what happened. Steps are left out by default, so ask for them.

run = client.get_process_data(process_data.processId, include_steps=True)
print(run.to_json())

If it failed and you've fixed the cause, send the plate back to the kitchen.

rerun_data = client.rerun(process_id=process_data.processId, version=1)
print(rerun_data.to_json())

A rerun runs the process again, so any step that sends an email or moves money can do it again. Check that your steps are safe to repeat first.

To stop or recover a batch, pass a list of process IDs.

# Cancel running orders.
response = client.bulk_terminate(process_ids=[process_data.processId])
print(response.details)

# Put failed or terminated orders back on the line.
response = client.bulk_resume(process_ids=[process_data.processId])
print(response.details)

There's also bulk_reviewed for marking failed or terminated runs as reviewed. And search_process_executions finds runs by name, status or correlation ID before you act on them.

The gotcha with max_in_progress

Here's the one that surprises people. The docs give a simple formula. Available tasks equal max in progress minus tasks currently in progress. A task counts as in progress from the moment the worker polls it until it submits the final result.

Now look at the numbers in this post. The worker allows 100 in progress, but the thread pool has 10 threads. A synchronous function runs up to ten at a time, and the rest wait in a memory queue. An async function that blocks on something slow causes the same pileup. A cook can hold a hundred tickets and still only have ten burners.

You have two ways out. Raise the thread pool or write async code that doesn't block. If your tasks finish quickly you may not need either, because a higher max_in_progress can be enough. You can also run more than one worker instance. A single poll request covers all registered workers, so a low poll interval is generally safe.

Where this fits

Unmeshed also has a built in Python step that runs a script inside the process itself. The SDK worker is for code that lives in your own application and runs wherever you start it.

That's the whole loop. Name a function, register it, add a step with the same name, run the process, and read the trail when something breaks.

If you've tuned a worker pool like this before, what bit you first? The thread pool or a call that quietly blocked?