Processing Documents

After a document arrives from the Peppol network, GoRoute records it as a transaction and notifies your webhook. This guide covers fetching that transaction's document, parsing it, and running it through your own processing pipeline.

Fetch the Document

Every received document belongs to a transaction, and the transaction's identifier is what you use to fetch it. The transaction.received webhook carries that identifier as transaction_id in its data object.

There are two endpoints, and the difference matters:

Endpoint	What it gives you	Use it for
GET /api/v1/transactions/{transaction_id}/document	The stored document as a download. When documents are held in object storage, this answers with JSON containing a short-lived download_url instead of the file itself.	Saving the original to your own archive
GET /api/v1/transactions/{transaction_id}/document/content	The raw XML, as text, served with the headers a browser needs to display it	Parsing in your own code, or showing the document on screen

Both need an API key with the transactions:read permission. Both answer 404 if the transaction does not exist, and also 404 if the transaction exists but no document was stored for it.

For parsing, /document/content is the one you want, because it hands you the XML directly:

import requests

API_BASE = "https://app.goroute.ai/peppol-api"


def fetch_document_xml(transaction_id: str) -> str:
    """Fetch the raw UBL XML of a received document."""
    response = requests.get(
        f"{API_BASE}/api/v1/transactions/{transaction_id}/document/content",
        headers={"X-API-Key": "your_api_key"},
    )
    response.raise_for_status()
    return response.text

If you are archiving rather than parsing, use the download endpoint and follow the pre-signed URL when one is returned:

def download_document(transaction_id: str) -> bytes:
    """Download the original stored document for a transaction."""
    response = requests.get(
        f"{API_BASE}/api/v1/transactions/{transaction_id}/document",
        headers={"X-API-Key": "your_api_key"},
    )
    response.raise_for_status()

    # Documents held in object storage come back as JSON with a short-lived link.
    if response.headers.get("Content-Type", "").startswith("application/json"):
        download_url = response.json()["download_url"]
        return requests.get(download_url).content

    return response.content

The document itself is XML, not a parsed object

GoRoute returns the document exactly as it arrived on the network — a UBL XML file. It is not returned as a pre-parsed JSON invoice with lines and totals broken out, and there is no endpoint that does that. Read the fields you need out of the XML yourself, as shown below. Earlier revisions of this page fetched a received document by a document identifier from the documents part of the API. No such endpoint has ever existed, and code written against it never worked. Fetch by transaction identifier, using the two endpoints above.

What Else the Transaction Tells You

The document is the invoice; the transaction record around it is the metadata — who sent it, what type it is, and when it arrived. Fetch it with GET /api/v1/transactions/{transaction_id}. The sender and receiver identifiers, the document type and the status also arrive in the webhook payload, so you often do not need a second call. See Tracking for the transaction record's full shape.

Parse UBL XML

from lxml import etree
from dataclasses import dataclass
from decimal import Decimal
from typing import Optional

@dataclass
class InvoiceLine:
    id: str
    description: str
    quantity: Decimal
    unit: str
    unit_price: Decimal
    line_total: Decimal
    tax_rate: Decimal

@dataclass
class Invoice:
    invoice_id: str
    issue_date: str
    due_date: Optional[str]
    currency: str
    seller_name: str
    seller_id: str
    buyer_name: str
    buyer_id: str
    subtotal: Decimal
    tax_amount: Decimal
    total: Decimal
    lines: list[InvoiceLine]

class UBLParser:
    NAMESPACES = {
        'inv': 'urn:oasis:names:specification:ubl:schema:xsd:Invoice-2',
        'cac': 'urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2',
        'cbc': 'urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2'
    }
    
    def parse_invoice(self, xml_content: str) -> Invoice:
        """Parse a UBL invoice XML string."""
        root = etree.fromstring(xml_content.encode())
        ns = self.NAMESPACES
        
        def xpath_text(path: str, default: str = "") -> str:
            result = root.xpath(path, namespaces=ns)
            return result[0].text if result else default
        
        # Parse invoice lines
        lines = []
        for line_elem in root.xpath('.//cac:InvoiceLine', namespaces=ns):
            lines.append(InvoiceLine(
                id=line_elem.xpath('cbc:ID/text()', namespaces=ns)[0],
                description=line_elem.xpath('.//cac:Item/cbc:Name/text()', namespaces=ns)[0],
                quantity=Decimal(line_elem.xpath('cbc:InvoicedQuantity/text()', namespaces=ns)[0]),
                unit=line_elem.xpath('cbc:InvoicedQuantity/@unitCode', namespaces=ns)[0],
                unit_price=Decimal(line_elem.xpath('.//cac:Price/cbc:PriceAmount/text()', namespaces=ns)[0]),
                line_total=Decimal(line_elem.xpath('cbc:LineExtensionAmount/text()', namespaces=ns)[0]),
                tax_rate=Decimal(line_elem.xpath('.//cac:ClassifiedTaxCategory/cbc:Percent/text()', namespaces=ns)[0] or '0')
            ))
        
        return Invoice(
            invoice_id=xpath_text('.//cbc:ID'),
            issue_date=xpath_text('.//cbc:IssueDate'),
            due_date=xpath_text('.//cbc:DueDate') or None,
            currency=xpath_text('.//cbc:DocumentCurrencyCode'),
            seller_name=xpath_text('.//cac:AccountingSupplierParty//cac:PartyLegalEntity/cbc:RegistrationName'),
            seller_id=xpath_text('.//cac:AccountingSupplierParty//cbc:EndpointID'),
            buyer_name=xpath_text('.//cac:AccountingCustomerParty//cac:PartyLegalEntity/cbc:RegistrationName'),
            buyer_id=xpath_text('.//cac:AccountingCustomerParty//cbc:EndpointID'),
            subtotal=Decimal(xpath_text('.//cac:LegalMonetaryTotal/cbc:LineExtensionAmount', '0')),
            tax_amount=Decimal(xpath_text('.//cac:TaxTotal/cbc:TaxAmount', '0')),
            total=Decimal(xpath_text('.//cac:LegalMonetaryTotal/cbc:PayableAmount', '0')),
            lines=lines
        )


# Usage
parser = UBLParser()
xml_content = fetch_document_xml("3fa85f64-5717-4562-b3fc-2c963f66afa6")
invoice = parser.parse_invoice(xml_content)

print(f"Invoice: {invoice.invoice_id}")
print(f"From: {invoice.seller_name}")
print(f"Total: {invoice.currency} {invoice.total}")

Attachments Arrive Inside the XML

A Peppol document does not have attachments hanging off it as separate files to be downloaded one by one. Anything attached to an invoice — a PDF rendering, a timesheet, a delivery note — is carried inside the UBL XML itself, base64-encoded in an cac:AdditionalDocumentReference/cac:Attachment/cbc:EmbeddedDocumentBinaryObject element. Once you have the XML you already have the attachment; decode it from the document you fetched.

import base64

def extract_attachments(xml_content: str) -> list[dict]:
    """Pull embedded attachments out of a received UBL document."""
    ns = UBLParser.NAMESPACES
    root = etree.fromstring(xml_content.encode())
    attachments = []

    for binary in root.xpath(
        './/cac:AdditionalDocumentReference//cbc:EmbeddedDocumentBinaryObject',
        namespaces=ns,
    ):
        attachments.append({
            "filename": binary.get("filename"),
            "mime_type": binary.get("mimeCode"),
            "content": base64.b64decode(binary.text or ""),
        })

    return attachments


for attachment in extract_attachments(xml_content):
    with open(attachment["filename"], "wb") as f:
        f.write(attachment["content"])

See Attachments for the same structure from the sending side, including the MIME types Peppol permits.

Processing Pipeline

Implement a robust processing pipeline:

from enum import Enum
import logging

class ProcessingStatus(Enum):
    RECEIVED = "received"
    VALIDATED = "validated"
    MATCHED = "matched"
    APPROVED = "approved"
    PAID = "paid"
    FAILED = "failed"

class DocumentProcessor:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.logger = logging.getLogger(__name__)
    
    async def process(self, transaction_id: str, document_type: str):
        """Process a received document through the pipeline.

        transaction_id and document_type both arrive in the transaction.received
        webhook payload, so no extra call is needed to start.
        """
        try:
            # 1. Fetch the document XML
            xml_content = await self.fetch_document_xml(transaction_id)
            self.logger.info(f"Processing transaction {transaction_id}: {document_type}")
            
            # 2. Parse the content
            if "Invoice" in document_type:
                parsed = self.parse_invoice(xml_content)
            elif "CreditNote" in document_type:
                parsed = self.parse_credit_note(xml_content)
            else:
                raise ValueError(f"Unsupported document type: {document_type}")
            
            # 3. Validate business rules
            self.validate_business_rules(parsed)
            
            # 4. Match to purchase order
            po_match = await self.match_to_po(parsed)
            
            # 5. Store in your system, keeping the original XML for audit
            await self.store_document(transaction_id, xml_content, parsed, po_match)
            
            # 6. Store any embedded attachments, which are already in the XML
            for attachment in self.extract_attachments(xml_content):
                await self.store_attachment(transaction_id, attachment)
            
            # 7. Update the status in your own system
            await self.update_status(transaction_id, ProcessingStatus.MATCHED)
            
            # 8. Trigger workflow
            await self.trigger_approval_workflow(transaction_id)
            
            return {"status": "success", "transaction_id": transaction_id}
            
        except Exception as e:
            self.logger.error(f"Failed to process {transaction_id}: {e}")
            await self.update_status(transaction_id, ProcessingStatus.FAILED)
            await self.notify_error(transaction_id, str(e))
            raise
    
    def validate_business_rules(self, invoice: Invoice):
        """Apply custom business validation."""
        errors = []
        
        # Check totals match
        calculated_total = invoice.subtotal + invoice.tax_amount
        if calculated_total != invoice.total:
            errors.append(f"Total mismatch: {calculated_total} != {invoice.total}")
        
        # Check due date is in future
        from datetime import date
        if invoice.due_date:
            due = date.fromisoformat(invoice.due_date)
            if due < date.today():
                errors.append(f"Due date {due} is in the past")
        
        if errors:
            raise ValueError(f"Business validation failed: {errors}")
    
    async def match_to_po(self, invoice: Invoice) -> dict:
        """Try to match invoice to a purchase order."""
        # Look for PO reference in invoice
        # This is business-specific logic
        return {"matched": False, "po_number": None}

The processing status in this example is yours, tracked in your own system. GoRoute records that the document was delivered to you; it does not hold a workflow state for what your accounts payable team has done with it.

Handle Different Document Types

The webhook payload's document_type is the full Peppol document type identifier, so match on a substring rather than an exact value.

async def handle_document(transaction_id: str, document_type: str):
    """Route a received document to the appropriate handler."""
    handlers = {
        "Invoice-2": handle_invoice,
        "CreditNote-2": handle_credit_note,
        "Order-2": handle_order,
        "OrderResponse-2": handle_order_response,
        "DespatchAdvice-2": handle_despatch_advice,
    }

    for marker, handler in handlers.items():
        if marker in document_type:
            return await handler(transaction_id)

    raise ValueError(f"No handler for {document_type}")

async def handle_invoice(transaction_id: str):
    """Process an incoming invoice."""
    invoice = parse_invoice(fetch_document_xml(transaction_id))
    
    # Create accounts payable record
    await create_ap_record(invoice)
    
    # Notify AP team
    await notify_team(
        "accounts_payable",
        f"New invoice received: {invoice.invoice_id} - {invoice.total}"
    )

async def handle_credit_note(transaction_id: str):
    """Process an incoming credit note."""
    credit_note = parse_credit_note(fetch_document_xml(transaction_id))
    
    # Find and update original invoice
    original = await find_invoice(credit_note.invoice_reference)
    if original:
        await apply_credit(original, credit_note)

Error Handling

class DocumentProcessingError(Exception):
    def __init__(self, transaction_id: str, message: str, recoverable: bool = True):
        self.transaction_id = transaction_id
        self.message = message
        self.recoverable = recoverable
        super().__init__(message)

async def process_with_error_handling(transaction_id: str):
    """Process a document with comprehensive error handling."""
    try:
        await process_document(transaction_id)
        
    except DocumentProcessingError as e:
        if e.recoverable:
            # Queue for retry
            await retry_queue.add(transaction_id)
            logger.warning(f"Transaction {transaction_id} queued for retry: {e.message}")
        else:
            # Send to manual review
            await manual_review_queue.add(transaction_id)
            logger.error(f"Transaction {transaction_id} requires manual review: {e.message}")
    
    except Exception as e:
        # Unexpected error - alert team
        logger.exception(f"Unexpected error processing {transaction_id}")
        await alert_team(f"Document processing failed: {transaction_id}")
        raise

A 404 from either document endpoint is worth handling separately. It means either that the transaction identifier is wrong, or that the transaction is real but no document was stored against it. Retrying will not change either, so send those to manual review rather than to the retry queue.

Telling GoRoute You Have Processed a Document

You do not, and there is no call to make. GoRoute's record of a received document ends at delivery: the document reached you, and the transaction says so. What happens next — approved, queried, paid, filed — lives in your system. Nothing on the GoRoute side is waiting for an acknowledgement from you, and nothing is left in a pending state if you never send one.

Replying to the supplier over Peppol is a different question, and is covered on Invoice Responses.

Best Practices

1. Process Asynchronously — Don't block webhook responses
2. Implement Idempotency — Handle duplicate deliveries, keyed on transaction_id
3. Validate Business Rules — Apply your own checks
4. Store Raw XML — Keep the original for audit, and for extracting attachments later
5. Log Everything — Detailed logs help debugging

Next Steps

• Responding to Invoices
• Webhook Setup
• Error Handling