get_ksef_invoice_status

Instructions

Implementation Reference

• src/mcp_ksef_pl/server.py:142-154 (handler)

  MCP tool handler for get_ksef_invoice_status — decorated with @mcp.tool, accepts a reference_number string and delegates to KSeFLifecycleManager.get_document_status.

  @mcp.tool
  async def get_ksef_invoice_status(reference_number: str) -> dict[str, Any]:
      """Retrieve the processing status of a submitted KSeF invoice (API v2).
  
      Parameters
      ----------
      reference_number: The 'reference_number' field from submit_invoice_to_ksef
                        ("{sessionRef}:{invoiceRef}").  Pass just the sessionRef
                        to retrieve the overall session status instead.
      """
      settings = KSeFSettings()
      manager = KSeFLifecycleManager(settings)
      return await manager.get_document_status(reference_number)

• src/mcp_ksef_pl/lifecycle.py:279-290 (helper)

  KSeFLifecycleManager.get_document_status — parses the compound reference (sessionRef:invoiceRef) or session-only reference and dispatches to KSeFClient.get_invoice_status or get_session_status.

  async def get_document_status(self, document_id: str) -> dict[str, Any]:
      """Get the status of a submitted invoice.
  
      Args:
          document_id: Either the compound "{sessionRef}:{invoiceRef}" string
                       returned by submit_document, or just a sessionRef to get
                       the overall session status.
      """
      if ":" in document_id:
          session_ref, invoice_ref = document_id.split(":", 1)
          return await self._client.get_invoice_status(session_ref, invoice_ref)
      return await self._client.get_session_status(document_id)

• src/mcp_ksef_pl/lifecycle.py:162-172 (helper)

  KSeFClient.get_invoice_status — makes a GET request to /sessions/{sessionRef}/invoices/{invoiceRef} and returns the JSON response (the actual KSeF API call to check invoice processing status).

  async def get_invoice_status(
      self,
      session_reference: str,
      invoice_reference: str,
  ) -> dict[str, Any]:
      """GET /sessions/{sessionRef}/invoices/{invoiceRef} — invoice status."""
      response = await self._request(
          "GET",
          f"/sessions/{session_reference}/invoices/{invoice_reference}",
      )
      return response.json()  # type: ignore[no-any-return]

• src/mcp_ksef_pl/lifecycle.py:162-172 (registration)

  Tool registration via @mcp.tool decorator on the get_ksef_invoice_status function (line 142 in server.py) using FastMCP framework — no separate registration file.

  async def get_invoice_status(
      self,
      session_reference: str,
      invoice_reference: str,
  ) -> dict[str, Any]:
      """GET /sessions/{sessionRef}/invoices/{invoiceRef} — invoice status."""
      response = await self._request(
          "GET",
          f"/sessions/{session_reference}/invoices/{invoice_reference}",
      )
      return response.json()  # type: ignore[no-any-return]