Inputs and timeouts

What goes into a workflow

Every workflow receives input parameters that define its execution context. Because these parameters travel over the network and are stored in the execution log, they must follow one strict rule: they must be JSON-serializable. Python primitive types pass without issue, as do lists, dictionaries, and null values.

# Primitive types
nom: str = "Marie"
age: int = 32
score: float = 0.95
actif: bool = True

# Collections
tags: list = ["urgent", "client-vip"]
metadata: dict = {"source": "api", "version": 2}

# Nullable
commentaire: str | None = None

However, datetime, bytes, set, tuple, and non-Pydantic custom classes are not supported directly. If your workflow needs to receive an invoice date, convert it to an ISO 8601 string before passing it in, and reconstruct the object inside the activity. This principle applies to all such types: conversion is your responsibility, not the platform’s.

Structure and validate with Pydantic

Passing a free-form dictionary works, but you lose all guarantees about its contents. For structured, validated parameters, rely on Pydantic models, which document the workflow interface as much as they protect it.

from pydantic import BaseModel, Field

class ParametresAnalyse(BaseModel):
    url_document: str
    langue: str = "fr"
    max_tokens: int = Field(default=1000, ge=100, le=10000)
    inclure_metadata: bool = False

@workflows.workflow.define(name="analyse_documentaire")
class AnalyseDocumentaire:
    @workflows.workflow.entrypoint
    async def run(self, params: ParametresAnalyse) -> dict:
        extraction = await extraire_texte(params.url_document, params.langue)
        analyse = await analyser_contenu(extraction["texte"], params.max_tokens)
        
        if params.inclure_metadata:
            metadata = await extraire_metadata(params.url_document)
            return {**analyse, "metadata": metadata}
        
        return analyse

On the caller side, there is nothing special to do: the fields of the BaseModel simply become top-level keys in the input dictionary.

execution = await client.workflows.execute_workflow_async(
    workflow_identifier="analyse_documentaire",
    input={
        "url_document": "https://example.com/rapport.pdf",
        "langue": "fr",
        "max_tokens": 2000,
        "inclure_metadata": True
    }
)

Pydantic automatically validates received data. If a field is invalid, the workflow fails immediately with an explicit error message, rather than starting execution and crashing three activities later. This validation tightens as needed: format constraints on an email address, numeric bounds on a priority, and above all extra = "forbid", which rejects unknown fields.

class ParametresStricts(BaseModel):
    email: str = Field(pattern=r"^[\w.+-]+@[\w-]+\.[\w.]+$")
    priorite: int = Field(ge=1, le=5)

    class Config:
        extra = "forbid"  # Rejects unknown fields

This last point deserves attention. Without extra="forbid", a typo in a field name goes unnoticed: the expected field takes its default value and the workflow runs with parameters that are not what you intended. The failure is silent, therefore costly.

When a single workflow must accept multiple forms of input, union types let you describe these variants without multiplying workflows.

from typing import Union

class TraitementTexte(BaseModel):
    texte: str
    type_traitement: str = "resume"

class TraitementURL(BaseModel):
    url: str
    type_traitement: str = "resume"

class ParametresFlexibles(BaseModel):
    source: Union[TraitementTexte, TraitementURL]

    class Config:
        extra = "forbid"

Execution lifetime

Every workflow has a maximum execution duration, set by the execution_timeout parameter. By default, it is one hour: past this deadline, the workflow is automatically canceled, regardless of its progress. For a process meant to last thirty seconds, this value is comfortable; for an overnight pipeline, it is a trap.

Adjust this timeout to the nature of the processing, both downward and upward. A fast-processing workflow benefits from a tight limit, because an overrun then signals a real problem rather than a temporary slowdown.

from datetime import timedelta

@workflows.workflow.define(
    name="traitement_rapide",
    execution_timeout=timedelta(minutes=10)
)
class TraitementRapide:
    @workflows.workflow.entrypoint
    async def run(self, donnees: dict) -> dict:
        return await traiter(donnees)

At the other end, the allowed ceiling is 7 days. This duration only makes sense for processes waiting for an external event, typically human validation: a report collected automatically but signed by a manager who will not return from leave until the following week.

from datetime import timedelta

@workflows.workflow.define(
    name="workflow_longue_duree",
    execution_timeout=timedelta(days=7)
)
class WorkflowLongueDuree:
    @workflows.workflow.entrypoint
    async def run(self, params: dict) -> dict:
        # This workflow can last up to 7 days
        # Useful for processes with human waiting periods
        donnees = await collecter_donnees_multi_sources(params)
        validation = await attendre_validation_humaine(donnees)
        return await finaliser_rapport(validation)

To calibrate, use this order of magnitude by usage family.

Two timeout levels not to confuse

execution_timeout bounds the entire workflow; start_to_close_timeout bounds a single activity. The two coexist and complement each other: the first protects you from a workflow that never completes, the second from an activity hanging on a silent service. Note in the example below that the HTTP client timeout (240 seconds) is intentionally lower than the start_to_close_timeout (5 minutes), so the error surfaces cleanly before the platform cuts the connection.

from datetime import timedelta

@workflows.activity(
    start_to_close_timeout=timedelta(minutes=5)
)
async def appel_api_externe(params: dict) -> dict:
    # This activity has 5 minutes to complete
    async with httpx.AsyncClient(timeout=240) as client:
        response = await client.post(params["url"], json=params["data"])
    return response.json()

@workflows.workflow.define(
    name="pipeline_multi_etapes",
    execution_timeout=timedelta(hours=2)
)
class PipelineMultiEtapes:
    @workflows.workflow.entrypoint
    async def run(self, sources: list) -> dict:
        # The entire workflow has 2 hours
        # But each individual activity has its own timeout
        resultats = []
        for source in sources:
            resultat = await appel_api_externe(source)
            resultats.append(resultat)
        return {"résultats": resultats}

One last reflex for your inputs: keep them light. The limit is 2 MB, and nothing forces you to approach it. Pass the URL of a document rather than its content, an order ID rather than the full order, and let the activities fetch the data. Define default values for everything optional, document each field with Field(description="..."), and entrust validation to Pydantic rather than if statements scattered throughout the workflow.

Key takeaways

• Workflow inputs must be JSON-serializable (str, int, float, bool, list, dict)
• Use Pydantic BaseModel for structured inputs with automatic validation
• The default timeout is 1 hour, configurable up to 7 days maximum
• Distinguish the workflow timeout (execution_timeout) from the activity timeout (start_to_close_timeout)
• Keep inputs light: IDs and URLs rather than large raw data