generate_fa3_invoice

Instructions

Implementation Reference

• src/mcp_ksef_pl/generator.py:423-523 (handler)

  FA3Generator class with the async generate() method that produces KSeF FA(3) XML. This is the core implementation invoked by the tool.

  class FA3Generator(BaseDocumentGenerator):
      """Generates KSeF FA(3) XML invoices for use with KSeF API v2.
  
      FA(3) is required for all new invoice submissions via KSeF API v2 online
      or batch sessions.  FA(2) is not accepted for new submissions.
  
      Schema: http://crd.gov.pl/wzor/2025/06/25/13775/
      XSD:    specs/schemat_FA(3)_v1-0E.xsd (reference copy; not bundled for validation)
      """
  
      def get_format_name(self) -> str:
          return "FA(3)"
  
      def get_country_code(self) -> str:
          return "PL"
  
      def get_namespace(self) -> str:
          return _NS3
  
      async def generate(self, invoice: InvoiceDocument) -> str:  # noqa: C901
          """Generate a KSeF-compliant FA(3) XML invoice.
  
          Produces a standard VAT invoice (RodzajFaktury=VAT).  Correction,
          advance, and settlement invoice types are not yet supported.
  
          Args:
              invoice: Structured invoice data.  seller.tax_id must be a Polish
                       NIP (10 digits).  buyer.tax_id may be a Polish NIP, a
                       EU VAT number, or absent (BrakID).
  
          Returns:
              UTF-8 FA(3) XML string ready to be passed to submit_invoice_to_ksef.
          """
          try:
              now_utc = _dt.datetime.now(_dt.UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
              vat_xml, p15_xml = _fa3_vat_fields(invoice.vat_summary or [])
              platnosc = _fa3_platnosc_block(invoice)
  
              parts: list[str] = [
                  '<?xml version="1.0" encoding="UTF-8"?>',
                  f'<Faktura xmlns="{_NS3}">',
                  # --- Naglowek ---
                  "  <Naglowek>",
                  '    <KodFormularza kodSystemowy="FA (3)" wersjaSchemy="1-0E">FA</KodFormularza>',
                  "    <WariantFormularza>3</WariantFormularza>",
                  f"    <DataWytworzeniaFa>{now_utc}</DataWytworzeniaFa>",
                  f"    <SystemInfo>{xml_escape(_SYSTEM_INFO)}</SystemInfo>",
                  "  </Naglowek>",
                  # --- Podmiot1 (seller) ---
                  f"  {_fa3_seller_block(invoice.seller).replace(chr(10), chr(10) + '  ').strip()}",
                  # --- Podmiot2 (buyer) ---
                  f"  {_fa3_buyer_block(invoice.buyer).replace(chr(10), chr(10) + '  ').strip()}",
                  # --- Fa ---
                  "  <Fa>",
                  f"    <KodWaluty>{xml_escape(invoice.currency)}</KodWaluty>",
                  f"    <P_1>{xml_escape(str(invoice.date))}</P_1>",
                  f"    <P_2>{xml_escape(invoice.number)}</P_2>",
              ]
  
              # VAT rate bands (only non-zero bands are emitted)
              if vat_xml:
                  for vl in vat_xml.splitlines():
                      parts.append(f"    {vl}")
  
              # Total gross (mandatory)
              parts.append(f"    {p15_xml}")
  
              # Adnotacje (mandatory, all defaults)
              for al in _fa3_adnotacje().splitlines():
                  parts.append(f"    {al}")
  
              # RodzajFaktury (mandatory — VAT for standard invoices)
              parts.append("    <RodzajFaktury>VAT</RodzajFaktury>")
  
              # Invoice lines (direct FaWiersz children, no wrapper)
              if invoice.lines:
                  for wl in _fa3_wiersz_lines(invoice).splitlines():
                      parts.append(f"    {wl}")
  
              # Payment block (optional)
              if platnosc:
                  for pl in platnosc.splitlines():
                      parts.append(f"    {pl}")
  
              parts.append("  </Fa>")
  
              # Stopka — optional note (correct location per XSD)
              if invoice.note:
                  parts += [
                      "  <Stopka>",
                      "    <Informacje>",
                      f"      <StopkaFaktury>{xml_escape(invoice.note)}</StopkaFaktury>",
                      "    </Informacje>",
                      "  </Stopka>",
                  ]
  
              parts.append("</Faktura>")
              return "\n".join(parts) + "\n"
  
          except Exception as exc:
              raise DocumentGenerationError(f"FA(3) generation failed: {exc}") from exc

• src/mcp_ksef_pl/server.py:62-76 (registration)

  The @mcp.tool decorated function 'generate_fa3_invoice' that registers the tool with FastMCP and delegates to FA3Generator.generate().

  @mcp.tool
  async def generate_fa3_invoice(invoice: InvoiceDocument) -> str:
      """Generate a KSeF-compliant FA(3) XML invoice from structured invoice data.
  
      FA(3) is required for all new invoice submissions via KSeF API v2.
      Use this tool — not generate_fa2_invoice — before calling submit_invoice_to_ksef.
  
      The seller's tax_id must be a Polish NIP (10 digits).
      The buyer's tax_id may be a Polish NIP, a EU VAT number (set alt_tax_id),
      or absent (leave tax_id.identifier empty to emit <BrakID>).
  
      Returns the FA(3) XML string ready for submit_invoice_to_ksef.
      """
      return await _fa3_generator.generate(invoice)

• src/mcp_ksef_pl/generator.py:209-353 (helper)

  FA(3) helper functions: _fa3_seller_block, _fa3_buyer_block, _fa3_vat_fields, _fa3_adnotacje, _fa3_wiersz_lines, _fa3_platnosc_block, and _adres_block — all called by FA3Generator.generate().

  def _adres_block(party: InvoiceParty) -> str:
      """Build a TAdres block (KodKraju + AdresL1 + optional AdresL2).
  
      TAdres in both FA(2) and FA(3) XSD contains only KodKraju, AdresL1, AdresL2,
      and GLN.  Structured postal/city fields are composed into AdresL1.
      """
      if not party.address:
          return ""
      a = party.address
  
      parts = [a.street or "", a.postal_code or "", a.city or ""]
      adres_l1 = " ".join(p for p in parts if p).strip(", ") or xml_escape(party.name or "")
  
      lines = [
          "<Adres>",
          f"  <KodKraju>{xml_escape(a.country_code.upper())}</KodKraju>",
          f"  <AdresL1>{xml_escape(adres_l1)}</AdresL1>",
      ]
      if a.province:
          # AdresL2 is optional — use for province/region when present
          lines.append(f"  <AdresL2>{xml_escape(a.province)}</AdresL2>")
      lines.append("</Adres>")
      return "\n".join(lines)
  
  
  def _fa3_seller_block(seller: InvoiceParty) -> str:
      """Build <Podmiot1> for FA(3)."""
      name = seller.name or f"{seller.first_name or ''} {seller.last_name or ''}".strip()
      nip = seller.tax_id.identifier if seller.tax_id.country_code.upper() == "PL" else ""
  
      id_lines = []
      if nip:
          id_lines.append(f"<NIP>{xml_escape(nip)}</NIP>")
      if seller.alt_tax_id:
          id_lines.append(f"<KodUE>{xml_escape(seller.tax_id.country_code.upper())}</KodUE>")
          id_lines.append(f"<NrVatUE>{xml_escape(seller.alt_tax_id.identifier)}</NrVatUE>")
      id_lines.append(f"<Nazwa>{xml_escape(name)}</Nazwa>")
  
      adres = _adres_block(seller)
  
      lines = ["<Podmiot1>", "  <DaneIdentyfikacyjne>"]
      for il in id_lines:
          lines.append(f"    {il}")
      lines.append("  </DaneIdentyfikacyjne>")
      if adres:
          for al in adres.splitlines():
              lines.append(f"  {al}")
      lines.append("</Podmiot1>")
      return "\n".join(lines)
  
  
  def _fa3_buyer_block(buyer: InvoiceParty) -> str:
      """Build <Podmiot2> for FA(3), including mandatory JST and GV flags.
  
      JST=2 means the invoice does not concern a subordinate local-government unit.
      GV=2  means the invoice does not concern a VAT-group member.
      Both default to 2 (not applicable) for standard B2B invoices.
      """
      name = buyer.name or f"{buyer.first_name or ''} {buyer.last_name or ''}".strip()
      nip = buyer.tax_id.identifier if buyer.tax_id.country_code.upper() == "PL" else ""
  
      id_lines: list[str] = []
      if nip:
          id_lines.append(f"<NIP>{xml_escape(nip)}</NIP>")
      elif buyer.alt_tax_id:
          id_lines.append(f"<KodUE>{xml_escape(buyer.tax_id.country_code.upper())}</KodUE>")
          id_lines.append(f"<NrVatUE>{xml_escape(buyer.alt_tax_id.identifier)}</NrVatUE>")
      else:
          id_lines.append("<BrakID>1</BrakID>")
      id_lines.append(f"<Nazwa>{xml_escape(name)}</Nazwa>")
  
      adres = _adres_block(buyer)
  
      lines = ["<Podmiot2>", "  <DaneIdentyfikacyjne>"]
      for il in id_lines:
          lines.append(f"    {il}")
      lines.append("  </DaneIdentyfikacyjne>")
      if adres:
          for al in adres.splitlines():
              lines.append(f"  {al}")
      lines.append("  <JST>2</JST>")
      lines.append("  <GV>2</GV>")
      lines.append("</Podmiot2>")
      return "\n".join(lines)
  
  
  def _fa3_vat_fields(summaries: list[VATSummary]) -> tuple[str, str]:
      """Return (vat_lines_xml, p15_xml) from the VAT summary list.
  
      The XSD groups (P_13_x, P_14_x) into optional inner sequences per rate band.
      Only bands with actual amounts are emitted.  P_15 (total gross) is always
      required.
      """
      band_lines: list[str] = []
      total_gross = Decimal("0")
  
      for s in summaries:
          rate_str = str(int(s.vat_rate)) if s.vat_rate == int(s.vat_rate) else str(s.vat_rate)
          idx = _VAT_RATE_FIELD.get(rate_str)
  
          if idx is not None:
              band_lines.append(f"<P_13_{idx}>{_d(s.taxable_base)}</P_13_{idx}>")
              if s.vat_amount > 0:
                  band_lines.append(f"<P_14_{idx}>{_d(s.vat_amount)}</P_14_{idx}>")
              total_gross += s.taxable_base + s.vat_amount
          elif s.vat_exemption_code:
              band_lines.append(f"<P_13_5>{_d(s.taxable_base)}</P_13_5>")
              total_gross += s.taxable_base
          else:
              band_lines.append(f"<P_13_1>{_d(s.taxable_base)}</P_13_1>")
              band_lines.append(f"<P_14_1>{_d(s.vat_amount)}</P_14_1>")
              total_gross += s.taxable_base + s.vat_amount
  
      vat_xml = "\n".join(band_lines)
      p15_xml = f"<P_15>{_d(total_gross)}</P_15>"
      return vat_xml, p15_xml
  
  
  def _fa3_adnotacje() -> str:
      """Return the mandatory <Adnotacje> block for a standard VAT invoice.
  
      All annotations default to 'not applicable' (2 / N values):
        P_16–P_18A, P_23  → 2 (not applicable for this invoice)
        Zwolnienie         → P_19N=1 (no VAT exemption)
        NoweSrodkiTransp.  → P_22N=1 (no new means of transport)
        PMarzy             → P_PMarzyN=1 (no margin scheme)
      """
      return (
          "<Adnotacje>\n"
          "  <P_16>2</P_16>\n"
          "  <P_17>2</P_17>\n"
          "  <P_18>2</P_18>\n"
          "  <P_18A>2</P_18A>\n"
          "  <Zwolnienie>\n"
          "    <P_19N>1</P_19N>\n"
          "  </Zwolnienie>\n"
          "  <NoweSrodkiTransportu>\n"
          "    <P_22N>1</P_22N>\n"
          "  </NoweSrodkiTransportu>\n"
          "  <P_23>2</P_23>\n"
          "  <PMarzy>\n"
          "    <P_PMarzyN>1</P_PMarzyN>\n"
          "  </PMarzy>\n"
          "</Adnotacje>"
      )

• src/mcp_ksef_pl/generator.py:28-44 (schema)

  Imports of InvoiceDocument and other core types that define the schema for the 'invoice' parameter of generate_fa3_invoice.

  from mcp_einvoicing_core import (
      BaseDocumentGenerator,
      DocumentGenerationError,
      InvoiceDocument,
      InvoiceParty,
      VATSummary,
      format_amount,
  )
  from mcp_einvoicing_core.xml_utils import xml_escape
  
  _NS = "http://crd.gov.pl/wzor/2023/06/29/12648/"
  _NS3 = "http://crd.gov.pl/wzor/2025/06/25/13775/"
  _SYSTEM_INFO = "mcp-ksef-pl/0.1.0"
  
  # Mapping from VAT rate (Decimal) to FA(2) P_13_x / P_14_x field index
  _VAT_RATE_FIELD: dict[str, int] = {
      "23": 1,