Flow Types

Flow Types#

class pywa.types.flows.FlowRequest#

Represents a flow request. This request is sent to the flow endpoint when a user interacts with a flow.

Variables:

  • version (str) – The version of the data_api_version specified on the flow json.

  • flow_token (str) – The flow token used to create the flow

  • action (pywa.types.flows.FlowRequestActionType) – The action that triggered the request.

  • screen (str | None) – The screen that triggered the request. If action is FlowRequestActionType.INIT or FlowRequestActionType.BACK, this field may be None.

  • data (dict[str, Any] | None) – The data sent from the screen. If action is FlowRequestActionType.BACK or FlowRequestActionType.INIT, this field may be None.

  • raw (dict[str, Any]) – The raw data of the request.

  • raw_encrypted (dict[str, str]) – The raw-encrypted data of the request.

respond(screen=None, data=None, error_message=None, close_flow=False, flow_token=None, version=None)#

Create a response for this request.

  • A shortcut for initializing a FlowResponse with the same version and flow token as this request.

Example

>>> wa = WhatsApp(business_private_key="...", ...)
>>> @wa.on_flow_request("/my-flow-endpoint")
... def my_flow_endpoint(_: WhatsApp, req: FlowRequest) -> FlowResponse:
...     return req.respond(
...         screen="SCREEN_ID",
...         data={"key": "value"},
...     )
Parameters:

  • screen (str | Screen | None) – The screen to display (if close_flow is False).

  • data (dict[str, str | int | float | bool | dict | DataSource | list[str | int | float | bool | dict | DataSource | NavigationItem]] | None) – The data to send to the screen or to add to flow completion .response dict (default to empty dict).

  • error_message (str | None) – This will redirect the user to screen and will trigger a snackbar error with the error_message present (if close_flow is False).

  • close_flow (bool) – Whether to close the flow or just navigate to the screen.

  • flow_token (str | None) – The flow token to close the flow (if close_flow is True, default to the request flow token).

  • version (str | None) – The version of the data exchange (Default to the request version).

Return type:

FlowResponse

token_no_longer_valid(error_message)#

Raise a FlowTokenNoLongerValid exception with the provided error message.

Example

>>> wa = WhatsApp(...)
>>> @wa.on_flow_request("/my-flow-endpoint")
... def my_flow_endpoint(_: WhatsApp, req: FlowRequest) -> FlowResponse:
...     if req.flow_token == "v1.0": # you see the token is no longer valid
...         req.token_no_longer_valid("Open the flow again to continue.")
...    ...
Parameters:

error_message (str) – The error message to display to the user.

Raises:

FlowTokenNoLongerValid – Well, always.

Return type:

NoReturn

property has_error: bool#

Check if the request has an error. When True, if flow endpoint register with acknowledge_errors=True, pywa will acknowledge the error and ignore the response from the callback. The callback still be called.

decrypt_media(key, index=0, dl_session=None)#

Decrypt the encrypted media file from the flow request.

Example

>>> from pywa import WhatsApp, types
>>> wa = WhatsApp(...)
>>> @wa.on_flow_request("/my-flow-endpoint")
... def my_flow_endpoint(_: WhatsApp, req: types.FlowRequest):
...     decrypted_data = req.decrypt_media(key="driver_license", index=0)
...     with open(decrypted_data.filename, "wb") as file:
...         file.write(decrypted_data.data)
...     return req.respond(...)
Parameters:

  • key (str) – The key of the media in the data (e.g. "driver_license").

  • index (int) – The index of the media in the data (default to 0).

  • dl_session (Client | None) – The HTTPX session to download the media (optional, new session will be created if not provided).

Returns:

  • media_id: The media ID of the decrypted media.

  • filename: The filename of the decrypted media.

  • decrypted_data: The decrypted data of the media.

Return type:

A tuple of (media_id, filename, decrypted_data) where

Raises:
class pywa.types.flows.FlowRequestActionType#

The type the action that triggered the FlowRequest.

Variables:

  • INIT – if the request is triggered when opening the flow (The FlowButton was sent with flow_action_type set to FlowActionType.DATA_EXCHANGE)

  • BACK – if the request is triggered when pressing back (The screen’s refresh_on_back attr set to True)

  • DATA_EXCHANGE – if the request is triggered by DataExchangeAction

  • NAVIGATE – if the FlowButton sent with FlowActionType.NAVIGATE and the screen is not in the routing_model (the request will contain an error)

class pywa.types.flows.FlowResponse#

Represents a flow response. This response is sent to the flow endpoint to determine the next screen to display or to close the flow. You should return this response from your flow endpoint callback.

Example

>>> from pywa import WhatsApp
>>> from pywa.types import FlowResponse

>>> wa = WhatsApp(business_private_key="...", ...)
>>> @wa.on_flow_request("/my-flow-endpoint")
... def my_flow_endpoint(_: WhatsApp, req: FlowRequest) -> FlowResponse:
...     return req.respond(  # shortcut for return FlowResponse(version=req.version, flow_token=req.flow_token, ...)
...         screen="SCREEN_ID",
...         data={"key": "value"},
...     )
Variables:
class pywa.types.flows.FlowCategory#

The category of the flow

Variables:

  • SIGN_UP – Sign up

  • SIGN_IN – Sign in

  • APPOINTMENT_BOOKING – Appointment booking

  • LEAD_GENERATION – Lead generation

  • CONTACT_US – Contact us

  • CUSTOMER_SUPPORT – Customer support

  • SURVEY – Survey

  • OTHER – Other

class pywa.types.flows.FlowDetails#

Represents the details of a flow.

Variables:

  • id (str) – The unique ID of the Flow.

  • name (str) – The user-defined name of the Flow which is not visible to users.

  • status (FlowStatus) – The status of the flow.

  • updated_at (datetime.datetime | None) – The last time the flow was updated (name, categories, endpoint_uri, json, etc.).

  • json_version (str | None) – The version specified by the developer in the Flow JSON asset uploaded.

  • data_api_version (str | None) – The version of the Data API specified by the developer in the Flow JSON asset uploaded. Only for Flows with an Endpoint.

  • categories (tuple[FlowCategory, ...]) – The categories of the flow.

  • validation_errors (tuple[FlowValidationError, ...] | None) – The validation errors of the flow (All errors must be fixed before the Flow can be published).

  • endpoint_uri (str | None) – The URL of the WA Flow Endpoint specified by the developer via API or in the Builder UI (Was data_channel_uri before v19.0).

  • preview (FlowPreview | None) – The URL to the web preview page to visualize the flow and its expiry time.

  • whatsapp_business_account (WhatsAppBusinessAccount | None) – The WhatsApp Business Account which owns the Flow.

  • application (FacebookApplication | None) – The Facebook developer application used to create the Flow initially.

  • health_status (dict | None) – A summary of the Flows health status.

publish()#
Update the status of this flow to FlowStatus.PUBLISHED.

  • This action is not reversible.

  • The Flow and its assets become immutable once published.

  • To update the Flow after that, you must create a new Flow. You specify the existing Flow ID as the clone_flow_id parameter while creating to copy the existing flow.

    You can publish your Flow once you have ensured that:

    • All validation errors and publishing checks have been resolved.

    • The Flow meets the design principles of WhatsApp Flows

    • The Flow complies with WhatsApp Terms of Service, the WhatsApp Business Messaging Policy and, if applicable, the WhatsApp Commerce Policy

Returns:

Whether the flow was published.

Raises:

FlowPublishingError – If this flow has validation errors or not all publishing checks have been resolved.

Return type:

SuccessResult

delete()#
When the flow is in FlowStatus.DRAFT status, you can delete it.
Returns:

Whether the flow was deleted.

Raises:

FlowDeletingError – If this flow is already published.

Return type:

SuccessResult

deprecate()#
When the flow is in FlowStatus.PUBLISHED status, you can only deprecate it.
Returns:

Whether the flow was deprecated.

Raises:

FlowDeprecatingError – If this flow is not published or already deprecated.

Return type:

SuccessResult

get_assets()#
Get all assets attached to this flow.
Returns:

The assets of the flow.

Return type:

Result[FlowAsset]

update_metadata(*, name=None, categories=None, endpoint_uri=None, application_id=None)#
Update the metadata of this flow.
Parameters:

  • name (str | None) – The name of the flow (optional).

  • categories (Iterable[FlowCategory | str] | None) – The new categories of the flow (optional).

  • endpoint_uri (str | None) – The URL of the FlowJSON Endpoint. Starting from FlowJSON 3.0 this property should be specified only gere. Do not provide this field if you are cloning a FlowJSON with version below 3.0.

  • application_id (int | None) – The ID of the Meta application which will be connected to the Flow. All the flows with endpoints need to have an Application connected to them.

Return type:

SuccessResult

Example

>>> from pywa.types.flows import FlowCategory
>>> wa = WhatsApp(waba_id='1234567890', ...)
>>> my_flows = wa.get_flows()
>>> my_flows[0].update_metadata(
...     name='Feedback',
...     categories=[FlowCategory.SURVEY, FlowCategory.OTHER],
...     endpoint_uri='https://my-api-server/feedback_flow'
... )
Returns:

Whether the flow was updated.

Raises:

ValueError – If neither name, categories or endpoint_uri is provided.

Return type:

SuccessResult

update_json(flow_json)#
Update the json of this flow.
Parameters:

flow_json (FlowJSON | dict | str | Path | bytes | BinaryIO) – The new json of the flow. Can be a FlowJSON object, dict, json str, json file path or json bytes.

Returns:

Whether the flow was updated.

Raises:

FlowUpdatingError – If the flow json is invalid or this flow is already published.

Return type:

FlowJSONUpdateResult

class pywa.types.flows.FlowStatus#

The status of the flow

Variables:

  • DRAFT – This is the initial status. The Flow is still under development. The Flow can only be sent with “mode”: “draft” for testing.

  • PUBLISHED – The Flow has been marked as published by the developer so now it can be sent to customers. This Flow cannot be deleted or updated afterwards.

  • DEPRECATED – The developer has marked the Flow as deprecated (since it cannot be deleted after publishing). This prevents sending and opening the Flow, to allow the developer to retire their endpoint. Deprecated Flows cannot be deleted or deprecated.

  • BLOCKED – Monitoring detected that the endpoint is unhealthy and set the status to Blocked. The Flow cannot be sent or opened in this state; the developer needs to fix the endpoint to get it back to Published state (more details in Flows Health and Monitoring).

  • THROTTLED –

    Monitoring detected that the endpoint is unhealthy and set the status to Throttled. Flows with throttled status can be opened, however only 10 messages of the Flow could be sent per hour. The developer needs to fix the endpoint to get it back to the PUBLISHED state (more details in Flows Health and Monitoring on developers.facebook.com).

class pywa.types.flows.FlowPreview#

Represents the preview of a flow.

Variables:

  • url (str) – The URL to the preview.

  • expires_at (datetime.datetime) – The expiration date of the preview.

with_params(interactive=None, flow_token=None, flow_action=None, flow_action_payload=None, phone_number=None, debug=None)#

Configure the interactive Web Preview.

Parameters:

  • interactive (bool | None) – If true, the preview will run in interactive mode. Defaults to false.

  • flow_token (str | None) – It will be sent as part of each request. You should always verify that token on your server to block any other unexpected requests. Required for Flows with endpoint.

  • flow_action (Literal[FlowRequestActionType.NAVIGATE, FlowRequestActionType.DATA_EXCHANGE] | None) – First action when Flow starts. data_exchange if it will make a request to the endpoint, or navigate if it won’t (this will also require flow_action_payload to be provided).

  • flow_action_payload (dict | None) – Initial screen data. Required if flow_action is navigate. Should be omitted otherwise.

  • phone_number (str | None) – Phone number that will be used to send the Flow, from which the public key will be used to encrypt the request payload. Required for Flows with endpoint.

  • debug (bool | None) – Show actions in a separate panel while interacting with the preview.

Returns:

The URL of the preview with the parameters added.

Return type:

str

class pywa.types.flows.FlowJSONUpdateResult#

Represents the result of updating a flow JSON.

Variables:
class pywa.types.flows.FlowValidationError#

Represents a validation error of a FlowJSON.

Variables:

  • error (str) – The error code.

  • error_type (str) – The type of the error.

  • message (str) – The error message.

  • line_start (int) – The start line of the error.

  • line_end (int) – The end line of the error.

  • column_start (int) – The start column of the error.

  • column_end (int) – The end column of the error.

class pywa.types.flows.FlowAsset#

Represents an asset in a flow.

Variables:

  • name (str) – The name of the asset (e.g. "flow.json").

  • type (str) – The type of the asset (e.g. "FLOW_JSON").

  • url (str) – The URL to the asset.

class pywa.types.flows.FlowMetricName#

The name of the metric

See Available Metrics.

Variables:

  • ENDPOINT_REQUEST_COUNT – Flow endpoint request count.

  • ENDPOINT_REQUEST_ERROR – Flow endpoint errors.

  • ENDPOINT_REQUEST_ERROR_RATE – Flow endpoint request error rate.

  • ENDPOINT_REQUEST_LATENCY_SECONDS_CEIL – Flow endpoint latency in seconds.

  • ENDPOINT_AVAILABILITY – Flow endpoint request error rate.

class pywa.types.flows.FlowMetricGranularity#

The granularity of the metric

Variables:

  • DAY – Daily granularity.

  • HOUR – Hourly granularity.

  • LIFETIME – Lifetime granularity.

class pywa.types.flows.CreatedFlow#

Represents a created flow.

Variables:

  • id (str) – The ID of the created flow.

  • success (bool) – Whether the flow json is valid (Only if created with flow json).

  • validation_errors (tuple[pywa.types.flows.FlowValidationError, ...] | None) – The validation errors of the flow json. Only available if success is False.

class pywa.types.flows.MigrateFlowsResponse#

Represents the response of the migrate flows endpoint.

Variables:
class pywa.types.flows.MigratedFlow#

Successfully migrated flow.

Variables:

  • source_id (str) – The ID of the flow in the source.

  • source_name (str) – The name of the flow in the source.

  • migrated_id (str) – The ID of the new flow in the destination.

class pywa.types.flows.MigratedFlowError#

Failed to migrate flow.

Variables:

  • source_name (str) – The name of the flow in the source.

  • error_code (str) – The error code.

  • error_message (str) – The error message.

class pywa.types.flows.FlowResponseError#

Bases: Exception

Base class for all flow response errors

  • Read more at developers.facebook.com.

  • Subclass this exception to return or raise from the flow endpoint callback (@wa.on_flow_request).

  • Override the status_code attribute to set the status code of the response.

  • Override the body attribute to set the body of the response (optional).

class pywa.types.flows.FlowTokenNoLongerValid#

Bases: FlowResponseError

This exception need to be returned or raised from the flow endpoint callback when the Flow token is no longer valid.

Example

>>> from pywa.types.flows import FlowTokenNoLongerValid
>>> wa = WhatsApp(...)
>>> @wa.on_flow_request("/my-flow-endpoint")
... def my_flow_endpoint(_: WhatsApp, req: FlowRequest) -> FlowResponse:
...     if req.flow_token == "v1.0":  # you see the token is no longer valid
...         # wa.send_message(..., buttons=FlowButton(...))  # resend the flow?
...         raise FlowTokenNoLongerValid(error_message='Open the flow again to continue.')
...    ...

  • The layout will be closed and the FlowButton will be disabled for the user.

  • You can send a new message to the user generating a new Flow token.

  • This action may be used to prevent users from initiating the same Flow again.

  • You are able to set an error message to display to the user. e.g. “The order has already been placed”

class pywa.types.flows.FlowRequestSignatureAuthenticationFailed#

Bases: FlowResponseError

This exception need to be returned or raised from the flow endpoint callback when the request signature authentication fails.

  • A generic error will be shown on the client.

class pywa.utils.FlowRequestDecryptedMedia#

Represents the decrypted media from a flow request.

Variables:

  • media_id (str) – The media ID.

  • filename (str) – The filename of the media.

  • data (bytes) – The decrypted media data.

pywa.utils.FlowRequestDecryptor#

Type hint for the function that decrypts the request from WhatsApp Flow.

Parameters:

  • encrypted_flow_data_b64 (str) – encrypted flow data

  • encrypted_aes_key_b64 (str) – encrypted AES key

  • initial_vector_b64 (str) – initial vector

  • private_key (str) – private key

  • password (str) – password for the private key. Optional.

Returns:

tuple[dict, bytes, bytes] - decrypted_data (dict): decrypted data from the request - aes_key (bytes): AES key you should use to encrypt the response - iv (bytes): initial vector you should use to encrypt the response

pywa.utils.FlowResponseEncryptor#

Type hint for the function that encrypts the response to WhatsApp Flow.

Parameters:

  • response (dict) – response to encrypt

  • aes_key (bytes) – AES key

  • iv (bytes) – initial vector

Returns:

encrypted response to send back to WhatsApp Flow

Return type:

encrypted_response (str)

pywa.utils.default_flow_request_decryptor(encrypted_flow_data_b64, encrypted_aes_key_b64, initial_vector_b64, private_key, password=None)#

The default global decryption function for decrypting data exchange requests from WhatsApp Flow.

  • This implementation follows the FlowRequestDecryptor type hint.

  • This implementation requires cryptography to be installed. To install it, run pip3 install ‘pywa[cryptography]’ or pip3 install cryptography.

  • This implementation was taken from the official documentation at developers.facebook.com.

Example

Set the default global decryptor (This is indeed the default):

>>> from pywa.utils import default_flow_request_decryptor
>>> from pywa import WhatsApp
>>> wa = WhatsApp(flows_request_decryptor=default_flow_request_decryptor, ...)

Set the decryptor for a specific flow:

>>> from pywa import WhatsApp
>>> from pywa.types.flows import FlowRequest, FlowResponse
>>> from pywa.utils import default_flow_request_decryptor
>>> wa = WhatsApp(...)
>>> @wa.on_flow_request("/sign-up-flow", request_decryptor=default_flow_request_decryptor)
... def on_sign_up_request(_: WhatsApp, flow: FlowRequest) -> FlowResponse | None: ...
Return type:

tuple[dict, bytes, bytes]

pywa.utils.default_flow_response_encryptor(response, aes_key, iv)#

The default global encryption function for encrypting data exchange responses to WhatsApp Flow.

  • This implementation follows the FlowResponseEncryptor type hint.

  • This implementation requires cryptography to be installed. To install it, run pip3 install ‘pywa[cryptography]’ or pip3 install cryptography.

  • This implementation was taken from the official documentation at developers.facebook.com.

Example

Set the default global encryptor (This is indeed the default):

>>> from pywa.utils import default_flow_response_encryptor
>>> from pywa import WhatsApp
>>> wa = WhatsApp(flows_response_encryptor=default_flow_response_encryptor, ...)

Set the encryptor for a specific flow:

>>> from pywa import WhatsApp
>>> from pywa.types.flows import FlowRequest, FlowResponse
>>> from pywa.utils import default_flow_response_encryptor
>>> wa = WhatsApp(...)
>>> @wa.on_flow_request("/sign-up-flow", response_encryptor=default_flow_response_encryptor)
... def on_sign_up_request(_: WhatsApp, flow: FlowRequest) -> FlowResponse | None: ...
Return type:

str

pywa.utils.flow_request_media_decryptor(encrypted_media, dl_session=None)#

Decrypt the encrypted media file from the flow request.

  • Read more at developers.facebook.com.

  • Use the .decrypt_media() shorthand method of the FlowRequest class instead.

  • This implementation requires cryptography to be installed. To install it, run pip3 install ‘pywa[cryptography]’ or pip3 install cryptography.

Example

>>> from pywa import WhatsApp, types
>>> wa = WhatsApp(...)
>>> @wa.on_flow_request("/media-upload")
... def on_media_upload_request(_: WhatsApp, req: types.FlowRequest) -> types.FlowResponse | None:
...     dec = req.decrypt_media(key="driver_license", index=0)
...     with open(dec.filename, "wb") as file:
...         file.write(dec.data)
...     return req.respond(...)
Parameters:

  • encrypted_media (dict) – encrypted media data from the flow request (see example above).

  • dl_session (httpx.Client) – download session. Optional.

Returns:

An object containing the media ID, filename, and decrypted data.

Raises:

  • HTTPStatusError – If the request to the CDN URL fails.

  • ValueError – If any of the hash verifications fail.

Return type:

FlowRequestDecryptedMedia

class pywa.handlers.FlowRequestCallbackWrapper#

This is a wrapper class for the flow request callback. It allows you to add more handlers to the same endpoint and split the logic into multiple functions.

  • THIS CLASS IS NOT MEANT TO BE INSTANTIATED DIRECTLY!

on(*, action, screen=None, filters=None)#

Decorator to help you add more handlers to the same endpoint and split the logic into multiple functions.

You can use this shortcuts: on_init(), on_data_exchange(), on_back().

Parameters:

  • action (FlowRequestActionType) – The action type to listen to.

  • screen (Screen | str | None) – The screen to listen to (if screen is not provided, the handler will be called for all screens for this action!).

  • filters (Filter) – A filter function to apply to the incoming request.

Returns:

The function itself.

Return type:

Callable[[_FlowRequestHandlerT], _FlowRequestHandlerT] | _FlowRequestHandlerT

on_back(*, screen=None, filters=None)#

Decorator to add a handler for the FlowRequestActionType.BACK action.

  • This request arrives when the user clicks the back button in the WhatsApp client and the flows screen refresh_on_back is set to True.

  • This callback must return a FlowResponse object, a dict or to raise FlowResponseError subclass.

Example

>>> wa = WhatsApp(...)

>>> @wa.on_flow_request("/feedback_flow")
>>> def feedback_flow_handler(_: WhatsApp, req: FlowRequest):
...     ...

>>> @feedback_flow_handler.on_back(screen="SURVEY")
>>> def on_back(_: WhatsApp, req: FlowRequest) -> FlowResponse:
...     return req.respond(...)
Parameters:

  • screen (Screen | str | None) – The screen to listen to (if screen is not provided, the handler will be called for all screens for this action!).

  • filters (Filter) – A filter function to apply to the incoming request.

Returns:

The callback function.

Return type:

Callable[[_FlowRequestHandlerT], _FlowRequestHandlerT]

on_data_exchange(screen=None, filters=None, *, call_on_error=False)#

Decorator to add a handler for the FlowRequestActionType.DATA_EXCHANGE action.

Example

>>> wa = WhatsApp(...)

>>> @wa.on_flow_request("/feedback_flow")
>>> def feedback_flow_handler(_: WhatsApp, req: FlowRequest):
...    ...

>>> @feedback_flow_handler.on_data_exchange(screen="SURVEY", filters=filters.new(lambda _, r: r.data["rating"] == "5"))
>>> def on_survey_with_rating_5(_: WhatsApp, req: FlowRequest) -> FlowResponse:
...     return req.respond(...)
Parameters:

  • screen (Screen | str | None) – The screen to listen to (if screen is not provided, the handler will be called for all screens for this action!).

  • filters (Filter) – A filter function to apply to the incoming request.

  • call_on_error (bool) – Whether to call the handler when the request has an error (The return value of the callback will be ignored).

Returns:

The callback function.

Return type:

Callable[[_FlowRequestHandlerT], _FlowRequestHandlerT]

on_init(filters=None, *, call_on_error=False)#

Decorator to add a handler for the FlowRequestActionType.INIT action.

  • This request arrives when the FlowButton sent with action_type of FlowRequestActionType.DATA_EXCHANGE.

Example

>>> wa = WhatsApp(...)

>>> wa.send_text(to=..., buttons=FlowButton(
...     # This will trigger the `INIT` request when the user clicks the button.
...     flow_action_type=FlowActionType.DATA_EXCHANGE, ...
... ))

>>> @wa.on_flow_request("/feedback_flow")
>>> def feedback_flow_handler(_: WhatsApp, req: FlowRequest):
...    ...

>>> @feedback_flow_handler.on_init  # This will be called when the flow is started.
>>> def on_init(_: WhatsApp, req: FlowRequest) -> FlowResponse:
...     return req.respond(...)
Parameters:

  • filters (Filter) – A filter function to apply to the incoming request.

  • call_on_error (bool) – Whether to call the handler when the request has an error (The return value of the callback will be ignored).

Returns:

The callback function.

Return type:

Callable[[_FlowRequestHandlerT], _FlowRequestHandlerT]