submit_flow

Instructions

Implementation Reference

• tools/flow_tools.py:46-157 (handler)

  The MCP tool handler for submit_flow. It is a @mcp.tool()-decorated async function that takes file_base64, file_name, flow_syntax, processing_rule, flow_type, and optional tracking_id. It decodes base64, validates it, calls the HTTP client (FlowClient.submit_flow), and returns the result.

  @mcp.tool()
  async def submit_flow(
      file_base64: Annotated[
          str,
          Field(
              description=(
                  "File content encoded in base64. "
                  "Accepted formats: Factur-X (PDF/A-3 with embedded XML), UBL 2.1 (XML), UN/CEFACT CII D22B (XML). "
                  "Maximum file size is defined by the Approved Platform (typically a few MB)."
              )
          ),
      ],
      file_name: Annotated[
          str,
          Field(
              description=(
                  "File name with extension (e.g. 'invoice_2024_001.xml', 'invoice_2024_001.pdf'). "
                  "The AP uses the extension to detect the format when flow_syntax is ambiguous."
              )
          ),
      ],
      flow_syntax: Annotated[
          str,
          Field(
              description=(
                  "Syntax/format of the submitted file (required). Common values: "
                  "FacturX — PDF/A-3 with embedded Factur-X XML; "
                  "UBL — UBL 2.1 XML invoice or credit note; "
                  "CII — UN/CEFACT CII D22B XML invoice; "
                  "CDAR — XML lifecycle status document; "
                  "EReporting — B2B or B2C e-reporting flow."
              )
          ),
      ],
      processing_rule: Annotated[
          ProcessingRule,
          Field(
              description=(
                  "Processing rule that determines routing and PPF transmission obligations. "
                  "B2B: domestic invoice between French VAT-registered entities (routed + reported to PPF). "
                  "B2BInt: international invoice or cross-border e-reporting. "
                  "B2C: invoice to a non-taxable entity or B2C e-reporting. "
                  "OutOfScope: transaction outside the reform scope (archived only). "
                  "ArchiveOnly: archiving without routing to recipient. "
                  "NotApplicable: used for lifecycle status (CDAR) flows."
              )
          ),
      ],
      flow_type: Annotated[
          str,
          Field(
              description=(
                  "Business type of the submitted flow. Common values: "
                  "Invoice, CreditNote, DebitNote, EReportingB2B, EReportingB2C, LifecycleStatus. "
                  "Refer to your Approved Platform's documentation for the exhaustive list."
              )
          ),
      ],
      tracking_id: Annotated[
          Optional[str],
          Field(
              default=None,
              description=(
                  "Sender-assigned tracking identifier (free-form, maxLength 36). "
                  "Recommended: use the invoice number or an internal UUID. "
                  "Allows retrieving this specific flow later via search_flows(tracking_id=...)."
              ),
          ),
      ] = None,
  ) -> dict:
      """
      Submit an electronic invoice, e-reporting, or lifecycle status to the Approved Platform.
  
      This is the primary action for sending B2B invoices (Factur-X, UBL, CII),
      B2BInt/B2C e-reportings, or CDAR lifecycle status messages.
  
      BEHAVIOR:
      - Submission is asynchronous: the AP returns a flowId and an initial status (typically 'Deposited'),
        not the final delivery status. Poll get_flow(flow_id) or search_flows to track processing.
      - Returns an error dict (with 'error' key) if the base64 encoding is invalid.
      - The AP may reject the flow synchronously (e.g. malformed XML, unknown recipient, quota exceeded);
        in that case the response contains an error code and message.
      - If processing_rule is B2B, the recipient must be registered in the PPF directory with an active
        directory line; verify with get_directory_line before submitting.
  
      RESPONSE on success: includes flowId (AP-assigned identifier), trackingId (echoed back),
      status (initial processing status), and submittedAt timestamp.
  
      USAGE GUIDELINES:
      - Always call get_directory_line (or search_directory_line) first to confirm the recipient is
        reachable and to identify their Approved Platform before submitting a B2B invoice.
      - Set a meaningful tracking_id (invoice number or UUID) to simplify later retrieval via search_flows.
      - After submission, use get_flow(flow_id, doc_type='Metadata') to monitor the flow status.
      - For lifecycle statuses on received invoices (Refused, Approved, etc.), prefer submit_lifecycle_status
        which provides structured status fields and handles mandatory PPF transmissions.
      - Call healthcheck_flow before a batch submission to confirm the AP is available.
      """
      try:
          file_content = base64.b64decode(file_base64)
      except Exception as e:
          return {"error": f"base64 decode failed: {e}"}
  
      client = get_flow_client()
      result = await client.submit_flow(
          file_content=file_content,
          file_name=file_name,
          flow_syntax=flow_syntax,
          processing_rule=processing_rule,
          flow_type=flow_type,
          tracking_id=tracking_id,
      )
      return result

• tools/flow_tools.py:43-44 (registration)

  The register_flow_tools function registers the submit_flow tool (and all other flow tools) on the FastMCP instance via the @mcp.tool() decorator at line 46.

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

• clients/flow_client.py:60-86 (handler)

  FlowClient.submit_flow — the actual HTTP client method that POSTs a multipart request to /v1/flows with the file content, flowInfo metadata, and optional tracking_id. This is called by the MCP handler.

  async def submit_flow(
      self,
      file_content: bytes,
      file_name: str,
      flow_syntax: str,
      processing_rule: Optional[ProcessingRule] = None,
      flow_type: Optional[str] = None,
      tracking_id: Optional[str] = None,
      sha256: Optional[str] = None,
  ) -> dict[str, Any]:
      """POST /v1/flows — Submit a flow (invoice, e-reporting, CDAR status)."""
      flow_info: dict[str, Any] = {"flowSyntax": flow_syntax, "name": file_name}
      if processing_rule:
          flow_info["processingRule"] = processing_rule
      if flow_type:
          flow_info["flowType"] = flow_type
      if tracking_id:
          flow_info["trackingId"] = tracking_id
      if sha256:
          flow_info["sha256"] = sha256
  
      files = {
          "file": (file_name, file_content, "application/octet-stream"),
          "flowInfo": (None, _json.dumps(flow_info), "application/json"),
      }
      response = await self._request("POST", "/v1/flows", files=files)
      return response.json()

• tests/test_mcp_protocol.py:201-268 (helper)

  MCP protocol tests for submit_flow covering: successful base64 decode + flowId return, invalid base64 error handling, and verifying decoded bytes + tracking_id are passed to the client.

  class TestFlowToolCalls:
      @pytest.mark.asyncio
      async def test_submit_flow_decodes_base64_and_returns_flow_id(self):
          """submit_flow decodes base64, calls the client, and returns the flowId."""
          fake_response = {"flowId": "FLOW-001", "trackingId": "TRK-001", "status": "Deposited"}
          mock_client = AsyncMock()
          mock_client.submit_flow = AsyncMock(return_value=fake_response)
  
          with patch("tools.flow_tools.get_flow_client", return_value=mock_client):
              async with Client(mcp) as client:
                  result = await client.call_tool(
                      "submit_flow",
                      {
                          "file_base64": base64.b64encode(b"<Invoice/>").decode(),
                          "file_name": "invoice.xml",
                          "flow_syntax": "CII",
                          "processing_rule": "B2B",
                          "flow_type": "Invoice",
                      },
                  )
  
          data = _parse(result)
          assert data["flowId"] == "FLOW-001"
          assert data["status"] == "Deposited"
  
      @pytest.mark.asyncio
      async def test_submit_flow_invalid_base64_returns_error_dict(self):
          """submit_flow returns {"error": ...} for invalid base64 without raising an MCP exception."""
          async with Client(mcp) as client:
              result = await client.call_tool(
                  "submit_flow",
                  {
                      "file_base64": "!!!not-valid-base64!!!",
                      "file_name": "invoice.xml",
                      "flow_syntax": "CII",
                      "processing_rule": "B2B",
                      "flow_type": "Invoice",
                  },
              )
  
          data = _parse(result)
          assert "error" in data
          assert "base64" in data["error"].lower()
  
      @pytest.mark.asyncio
      async def test_submit_flow_passes_decoded_bytes_to_client(self):
          """submit_flow passes decoded bytes (not the base64 string) to the HTTP client."""
          xml_bytes = b"<Invoice><ID>001</ID></Invoice>"
          mock_client = AsyncMock()
          mock_client.submit_flow = AsyncMock(return_value={"flowId": "F-001", "status": "Deposited"})
  
          with patch("tools.flow_tools.get_flow_client", return_value=mock_client):
              async with Client(mcp) as client:
                  await client.call_tool(
                      "submit_flow",
                      {
                          "file_base64": base64.b64encode(xml_bytes).decode(),
                          "file_name": "invoice.xml",
                          "flow_syntax": "CII",
                          "processing_rule": "B2B",
                          "flow_type": "Invoice",
                          "tracking_id": "TRK-042",
                      },
                  )
  
          call_kwargs = mock_client.submit_flow.call_args.kwargs
          assert call_kwargs["file_content"] == xml_bytes
          assert call_kwargs["tracking_id"] == "TRK-042"

• server.py:63-63 (registration)

  server.py calls register_flow_tools(mcp) which registers the submit_flow tool (and others) via the fastmcp decorator.

  register_flow_tools(mcp)