Anatomy of a Marqov Script

This guide covers how to structure a Marqov script so the platform executor can discover and run it.

How the Executor Finds Your Entry Point

When you submit a script, the Temporal executor (temporal_executor.py) runs it via exec() and then looks for an entry point using two strategies, tried in order:

main(client, params) — an explicit async function you define
Single @workflow function — auto-detected via the _is_workflow attribute

If neither is found, or if multiple @workflow functions exist without a main(), the executor raises a RuntimeError.

Strategy 1: Explicit `main(client, params)`

Define an async function named main that receives the Temporal client and the job parameters dict. This gives you full control over dispatch, including running multiple workflows or doing pre/post-processing.


from marqov import task, workflow, Circuit
 
@task(retries=2, timeout=600)
async def run_circuit(circuit_dict: dict, backend_config: dict) -> dict:
    circuit = Circuit.from_dict(circuit_dict)
    # ... execute circuit ...
    return {"counts": {"00": 500, "11": 500}}
 
@workflow(name="Bell-State")
def bell_experiment(backend_config: dict):
    circuit = Circuit().h(0).cnot(0, 1)
    return run_circuit(circuit.to_dict(), backend_config)
 
async def main(client, params):
    """Explicit entry point -- full control over dispatch."""
    dispatch = bell_experiment(backend_config=params)
    result = await dispatch.run(client)
    return {
        "result": result,
        "_summary": {
            "State |00>": str(result["counts"].get("00", 0)),
            "State |11>": str(result["counts"].get("11", 0)),
        },
    }

The executor calls main(temporal_client, params_dict) directly. If it returns a dict, that dict is stored as the job result. If it returns any other type, the executor wraps it as {"result": value}.

Strategy 2: Single `@workflow` Function

If your script defines exactly one @workflow-decorated function and no main(), the executor auto-detects it and dispatches it. The executor filters the job params dict to match the workflow function’s signature (see Parameter Handling for details).


from marqov import task, workflow, Circuit
 
@task(retries=2, timeout=600)
async def run_circuit(circuit_dict: dict, backend_config: dict) -> dict:
    circuit = Circuit.from_dict(circuit_dict)
    # ... execute circuit ...
    return {"counts": {"00": 500, "11": 500}}
 
@workflow(name="Bell-State")
def bell_experiment(backend_config: dict):
    """Single workflow -- auto-detected by executor."""
    circuit = Circuit().h(0).cnot(0, 1)
    return run_circuit(circuit.to_dict(), backend_config)

The executor calls bell_experiment(**filtered_params) automatically, then uses dispatch.run_with_ids(client) to execute and retrieve workflow/run IDs for tracking.

When to Use Which Strategy

Scenario	Strategy
Simple single-workflow script	Single `@workflow` (less boilerplate)
Need pre/post-processing around dispatch	`main(client, params)`
Need to return `_summary` cards for the dashboard	`main(client, params)`
Multiple workflows in one script	`main(client, params)` (required)
Need to call `dispatch()` (fire-and-forget)	`main(client, params)`

Common Mistakes

Multiple @workflow functions without main():


@workflow
def workflow_a():
    ...
 
@workflow
def workflow_b():
    ...
 
# ERROR: "Script defines 2 @workflow functions (workflow_a, workflow_b).
#         Define a main(client, params) function to specify which to run."

Fix this by adding a main() that picks which workflow to run:


async def main(client, params):
    if params.get("mode") == "a":
        dispatch = workflow_a()
    else:
        dispatch = workflow_b()
    return await dispatch.run(client)

No entry point at all:


from marqov import task
 
@task
def add(x, y):
    return x + y
 
# ERROR: "Script must define either main(client, params) or
#         a single @workflow function."

You need at least one @workflow function or a main().

Minimal Complete Examples

Minimal `main()` script


from marqov import task, workflow
 
@task
def add(x, y):
    return x + y
 
@workflow
def compute():
    return add(1, 2)
 
async def main(client, params):
    dispatch = compute()
    result = await dispatch.run(client)
    return {"result": result}

Minimal auto-detected script


from marqov import task, workflow
 
@task
def add(x, y):
    return x + y
 
@workflow
def compute():
    return add(1, 2)