submit_lifecycle_status

Instructions

Implementation Reference

• tools/flow_tools.py:335-440 (handler)

  MCP tool handler for submit_lifecycle_status. Decorated with @mcp.tool() inside register_flow_tools(). Validates read-only mode, checks flow not in terminal state, requires user confirmation via ConfirmationGate, then delegates to FlowClient.submit_lifecycle_status.

  async def submit_lifecycle_status(
      referenced_flow_id: Annotated[
          str,
          Field(
              description=(
                  "Identifier of the invoice flow to which this status applies "
                  "(flowId returned upon receipt, maxLength 36)."
              )
          ),
      ],
      status_code: Annotated[
          LifecycleStatusCode,
          Field(
              description=(
                  "Lifecycle status code to emit. Values defined in XP Z12-014: "
                  "Refused (transmitted to PPF), "
                  "Approved, "
                  "PartiallyApproved, "
                  "Disputed, "
                  "Suspended, "
                  "Cashed (transmitted to PPF), "
                  "PaymentTransmitted, "
                  "Cancelled. "
                  "Refused and Cashed are mandatory transmissions to PPF."
              )
          ),
      ],
      reason: Annotated[
          Optional[str],
          Field(
              default=None,
              description=(
                  "Status reason, mandatory for Refused and Disputed. "
                  "Free text describing the reason for refusal or dispute."
              ),
          ),
      ] = None,
      payment_date: Annotated[
          Optional[str],
          Field(
              default=None,
              description=(
                  "Payment date (ISO 8601 format: YYYY-MM-DD). "
                  "Provided for Cashed and PaymentTransmitted statuses."
              ),
          ),
      ] = None,
      payment_amount: Annotated[
          Optional[str],
          Field(
              default=None,
              description=(
                  "Payment amount (decimal string, e.g. '1250.00'). "
                  "Provided for Cashed and PaymentTransmitted statuses."
              ),
          ),
      ] = None,
      confirmation_token: Annotated[
          Optional[str],
          Field(
              default=None,
              description=(
                  "Confirmation token from a previous call. "
                  "Omit on the first call; supply on the second call to execute."
              ),
          ),
      ] = None,
  ) -> dict:
      """
      Emit a processing status on a received invoice: Refused, Approved,
      PartiallyApproved, Disputed, Suspended, Cashed, PaymentTransmitted,
      Cancelled. Refused and Cashed are mandatory transmissions to PPF.
      Reason is mandatory for Refused and Disputed.
  
      HUMAN-IN-THE-LOOP: Requires user confirmation. Call without confirmation_token
      first, show the summary to the user, then call again with the token.
      """
      assert_not_read_only("FR_READ_ONLY")
      client = get_flow_client()
      try:
          await _check_flow_not_terminal(referenced_flow_id, client)
      except ValueError as exc:
          return {"error": str(exc)}
  
      gate = ConfirmationGate.get_default()
      if not gate.is_confirmed(confirmation_token):
          reason_note = f", reason={reason!r}" if reason else ""
          return gate.pending_response(
              action="submit_lifecycle_status",
              summary=(
                  f"Emit lifecycle status {status_code!r} on flow "
                  f"{referenced_flow_id!r}{reason_note}. "
                  "Refused and Cashed statuses are transmitted to PPF and cannot be retracted."
              ),
              token=confirmation_token,
          )
  
      result = await client.submit_lifecycle_status(
          referenced_flow_id=referenced_flow_id,
          status_code=status_code,
          reason=reason,
          payment_date=payment_date,
          payment_amount=payment_amount,
      )
      gate.consume(confirmation_token)
      return result

• clients/flow_client.py:112-139 (handler)

  HTTP client method that builds CDAR lifecycle status XML via _build_lifecycle_status_xml helper and POSTs it as a multipart form (file + flowInfo JSON) to the AP's /v1/flows endpoint.

  async def submit_lifecycle_status(
      self,
      referenced_flow_id: str,
      status_code: LifecycleStatusCode,
      reason: Optional[str] = None,
      payment_date: Optional[str] = None,
      payment_amount: Optional[str] = None,
  ) -> dict[str, Any]:
      """POST /v1/flows — Submit a CDAR lifecycle status."""
      status_xml = _build_lifecycle_status_xml(
          referenced_flow_id=referenced_flow_id,
          status_code=status_code,
          reason=reason,
          payment_date=payment_date,
          payment_amount=payment_amount,
      )
      flow_info: dict[str, Any] = {
          "flowSyntax": "CDAR",
          "processingRule": "NotApplicable",
          "flowType": "SupplierInvoiceLC",
          "name": "lifecycle_status.xml",
      }
      files = {
          "file": ("lifecycle_status.xml", status_xml.encode("utf-8"), "application/xml"),
          "flowInfo": (None, _json.dumps(flow_info), "application/json"),
      }
      response = await self._request("POST", "/v1/flows", files=files)
      return response.json()

• clients/flow_client.py:204-229 (helper)

  Helper that builds the CDAR lifecycle status XML payload per XP Z12-014. Includes optional Reason and Payment elements (Date, Amount) in the XML structure.

  def _build_lifecycle_status_xml(
      referenced_flow_id: str,
      status_code: str,
      reason: Optional[str] = None,
      payment_date: Optional[str] = None,
      payment_amount: Optional[str] = None,
  ) -> str:
      """Build a CDAR lifecycle status XML document (XP Z12-014)."""
      reason_el = f"<Reason>{_xml_escape(reason)}</Reason>" if reason else ""
      payment_el = ""
      if payment_date or payment_amount:
          payment_el = (
              "<Payment>"
              + (f"<Date>{_xml_escape(payment_date)}</Date>" if payment_date else "")
              + (f"<Amount>{_xml_escape(payment_amount)}</Amount>" if payment_amount else "")
              + "</Payment>"
          )
      return (
          '<?xml version="1.0" encoding="UTF-8"?>'
          '<LifecycleStatus xmlns="urn:xp-z12-013:lifecycle-status:1.0">'
          f"<ReferencedFlowId>{_xml_escape(referenced_flow_id)}</ReferencedFlowId>"
          f"<StatusCode>{_xml_escape(status_code)}</StatusCode>"
          f"{reason_el}"
          f"{payment_el}"
          "</LifecycleStatus>"
      )

• clients/flow_client.py:40-49 (schema)

  LifecycleStatusCode type definition — a Literal union of all valid CDAR status values (Refused, Approved, PartiallyApproved, Disputed, Suspended, Cashed, PaymentTransmitted, Cancelled). Used as the type for the status_code parameter.

  LifecycleStatusCode = Literal[
      "Refused",
      "Approved",
      "PartiallyApproved",
      "Disputed",
      "Suspended",
      "Cashed",
      "PaymentTransmitted",
      "Cancelled",
  ]

• tools/flow_tools.py:65-66 (registration)

  Registration function that wraps all flow tools. Within this function, submit_lifecycle_status is registered via @mcp.tool() decorator. Called from server.py line 74: register_flow_tools(mcp).

  def register_flow_tools(mcp: FastMCP) -> None:
      """Registers the 5 Flow Service tools on the FastMCP instance."""