Building a Scalable Contract Data Extraction Pipeline with Microsoft Foundry and Python

Architecture Overview

Alt text: Architecture diagram showing Blob Storage triggering Azure Function, calling Document Intelligence, transforming data, and storing in Cosmos DB

Flow:

• Upload contract files (PDF or ZIP) to Azure Blob Storage
• Azure Function triggers automatically on file upload
• Azure AI Document Intelligence extracts layout and tables
• A transformation layer converts output into a canonical JSON format
• Data is stored in Azure Cosmos DB

Step 1: Trigger Processing with Azure Functions

An Azure Function with a Blob trigger enables automatic processing when a file is uploaded.

import logging
import azure.functions as func
import zipfile
import io

def main(myblob: func.InputStream):
    logging.info(f"Processing blob: {myblob.name}")

    if myblob.name.endswith(".zip"):
        with zipfile.ZipFile(io.BytesIO(myblob.read())) as z:
            for file_name in z.namelist():
                logging.info(f"Extracting {file_name}")
                file_data = z.read(file_name)
                # Pass file_data to extraction step

Best Practices

• Keep functions stateless and idempotent
• Handle retries for transient failures
• Store configuration in environment variables

Step 2: Extract Layout Using Document Intelligence

The prebuilt layout model helps extract tables, text, and structure from documents.

from azure.ai.documentintelligence import DocumentIntelligenceClient
from azure.core.credentials import AzureKeyCredential

client = DocumentIntelligenceClient(
    endpoint="<your-endpoint>",
    credential=AzureKeyCredential("<your-key>")
)

poller = client.begin_analyze_document(
    "prebuilt-layout",
    document=file_data
)

result = poller.result()

Output Includes

• Structured tables
• Paragraphs and text blocks
• Bounding regions for layout context

Step 3: Handle Multi-Page Table Continuity

Contract documents often contain tables split across multiple pages. These need to be merged to preserve data integrity.

def merge_tables(tables):
    merged = []
    current = None

    for table in tables:
        headers = [cell.content for cell in table.cells if cell.row_index == 0]

        if current and headers == current["headers"]:
            current["rows"].extend(extract_rows(table))
        else:
            if current:
                merged.append(current)
            current = {
                "headers": headers,
                "rows": extract_rows(table)
            }

    if current:
        merged.append(current)

    return merged

Key Considerations

• Match headers to detect continuation
• Preserve row order
• Avoid duplicate headers

Step 4: Transform to a Canonical JSON Schema

A consistent schema ensures compatibility across downstream systems.

{
  "id": "contract_123",
  "documentType": "contract",
  "vendorName": "ABC Corp",
  "invoiceDate": "2023-05-05",
  "tables": [
    {
      "name": "Line Items",
      "headers": ["Item", "Qty", "Price"],
      "rows": [
        ["Service A", "2", "100"]
      ]
    }
  ],
  "metadata": {
    "sourceFile": "contract.pdf",
    "processedAt": "2026-04-22T10:00:00Z"
  }
}

Design Tips

• Keep schema flexible and extensible
• Include metadata for traceability
• Avoid excessive nesting

Step 5: Persist Data in Cosmos DB

Store the transformed data in a scalable NoSQL database.

from azure.cosmos import CosmosClient

client = CosmosClient("<cosmos-uri>", "<key>")
database = client.get_database_client("contracts-db")
container = database.get_container_client("documents")

container.upsert_item(canonical_json)

Best Practices

• Choose an appropriate partition key (for example, documentType or vendorName)
• Optimize indexing policies
• Monitor request units (RU) usage

Observability and Monitoring

To ensure reliability:

• Enable logging with Application Insights
• Track processing time and failures
• Monitor document extraction accuracy

Security Considerations

• Store secrets securely using Azure Key Vault
• Use Managed Identity for service authentication
• Apply role-based access control (RBAC) to storage resources

Conclusion

This approach provides a scalable and maintainable solution for contract data extraction:

• Event-driven processing with Azure Functions
• Accurate extraction using Document Intelligence
• Clean transformation into a reusable schema
• Efficient storage with Cosmos DB
• This foundation can be extended with validation layers, review workflows, or analytics dashboards depending on your business requirements.

Resources

Contract data extraction – Document Intelligence: Foundry Tools | Microsoft Learn
microsoft/content-processing-solution-accelerator: Programmatically extract data and apply schemas to unstructured documents across text-based and multi-modal content using Azure AI Foundry, Azure OpenAI, Azure AI Content Understanding, and Cosmos DB.