How to Integrate Azure OpenAI for payments with CosmosDB for startups

Azure OpenAI gives you the reasoning layer for payment workflows: classify intents, extract fields from invoices, flag suspicious transactions, and generate customer-facing responses. CosmosDB gives you the durable state layer: store payment sessions, retries, audit trails, and conversation memory without fighting schema migrations.

For startups building AI agents around payments, this combo is practical. You get an agent that can read a payment request, decide what to do next, and persist every decision for compliance and recovery.

Prerequisites

• •Python 3.10+
• •An Azure subscription
• •An Azure OpenAI resource with:
  • •endpoint
  • •api_key
  • •a deployed model name like gpt-4o-mini
• •An Azure Cosmos DB account with:
  • •endpoint
  • •key
  • •database and container created
• •pip installed
• •Environment variables set:
  • •AZURE_OPENAI_ENDPOINT
  • •AZURE_OPENAI_API_KEY
  • •AZURE_OPENAI_DEPLOYMENT_NAME
  • •COSMOS_ENDPOINT
  • •COSMOS_KEY
  • •COSMOS_DATABASE_NAME
  • •COSMOS_CONTAINER_NAME

Install the SDKs:

pip install openai azure-cosmos python-dotenv

Integration Steps

1) Initialize Azure OpenAI and CosmosDB clients

Use the Azure OpenAI chat client for payment reasoning, and CosmosDB for persistence. Keep them in separate modules in real projects; here they live together so you can see the flow end to end.

import os
from dotenv import load_dotenv
from openai import AzureOpenAI
from azure.cosmos import CosmosClient, PartitionKey

load_dotenv()

aoai_client = AzureOpenAI(
    api_key=os.environ["AZURE_OPENAI_API_KEY"],
    api_version="2024-06-01",
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
)

cosmos_client = CosmosClient(
    url=os.environ["COSMOS_ENDPOINT"],
    credential=os.environ["COSMOS_KEY"],
)

database_name = os.environ["COSMOS_DATABASE_NAME"]
container_name = os.environ["COSMOS_CONTAINER_NAME"]

database = cosmos_client.create_database_if_not_exists(id=database_name)
container = database.create_container_if_not_exists(
    id=container_name,
    partition_key=PartitionKey(path="/payment_id"),
)

2) Define the payment extraction prompt

The agent should turn raw payment text into structured JSON. That makes it easy to validate fields before writing anything to CosmosDB.

import json

def extract_payment_details(raw_text: str) -> dict:
    response = aoai_client.chat.completions.create(
        model=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
        messages=[
            {
                "role": "system",
                "content": (
                    "You extract structured payment data from user messages. "
                    "Return only valid JSON with keys: "
                    "payment_id, payer_name, amount, currency, due_date, "
                    "merchant, confidence."
                ),
            },
            {"role": "user", "content": raw_text},
        ],
        temperature=0,
    )

    content = response.choices[0].message.content
    return json.loads(content)

3) Store the extracted payment record in CosmosDB

Use CosmosDB as your source of truth for payment workflow state. A simple pattern is to write an initial record with status pending_review, then update it after downstream checks.

from datetime import datetime, timezone

def save_payment_record(payment: dict) -> dict:
    item = {
        "id": payment["payment_id"],
        "payment_id": payment["payment_id"],
        "payer_name": payment["payer_name"],
        "amount": payment["amount"],
        "currency": payment["currency"],
        "due_date": payment["due_date"],
        "merchant": payment["merchant"],
        "confidence": payment["confidence"],
        "status": "pending_review",
        "created_at": datetime.now(timezone.utc).isoformat(),
    }

    container.upsert_item(item)
    return item

4) Add an approval or risk check using Azure OpenAI

This is where the agent becomes useful for payments. Ask the model to review the structured data and decide whether it looks safe enough to proceed or whether it needs manual review.

def assess_payment_risk(payment: dict) -> dict:
    response = aoai_client.chat.completions.create(
        model=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a payments risk assistant. "
                    "Return only valid JSON with keys: decision, reason."
                ),
            },
            {
                "role": "user",
                "content": f"Assess this payment for risk:\n{json.dumps(payment)}",
            },
        ],
        temperature=0,
    )

    return json.loads(response.choices[0].message.content)

5) Update CosmosDB with the final workflow state

Once the agent decides what to do next, persist that decision. This gives you auditability and lets other services pick up the record later.

def process_payment_request(raw_text: str):
    payment = extract_payment_details(raw_text)
    saved = save_payment_record(payment)
    risk = assess_payment_risk(saved)

    saved["status"] = risk["decision"]
    saved["risk_reason"] = risk["reason"]
    saved["updated_at"] = datetime.now(timezone.utc).isoformat()

    container.upsert_item(saved)
    return saved

result = process_payment_request(
    "Invoice INV-2048 from Acme Supplies for $420 USD due on 2026-05-01"
)
print(result)

Testing the Integration

Run a simple end-to-end test with a known input. You want to verify three things:

• •Azure OpenAI returns valid JSON
• •CosmosDB accepts the record
• •The stored status updates after risk assessment

test_input = "Payment request PMT-1007 for Jane Doe paying $125 USD to CloudBooks due on 2026-04-30"
result = process_payment_request(test_input)

print("Stored record:")
print(json.dumps(result, indent=2))

stored_item = container.read_item(item=result["payment_id"], partition_key=result["payment_id"])
print("\nFetched from CosmosDB:")
print(json.dumps(stored_item, indent=2))

Expected output:

Stored record:
{
  "id": "PMT-1007",
  "payment_id": "PMT-1007",
  "payer_name": "Jane Doe",
  "amount": 125,
  "currency": "USD",
  "due_date": "...",
  "merchant": "CloudBooks",
  ...
}

Fetched from CosmosDB:
{
  ...
  "status": "...",
  ...
}

If your model returns malformed JSON, fix that first by tightening the system prompt or adding a JSON schema validation layer before writing to CosmosDB.

Real-World Use Cases

• •

  Invoice intake agent

  • •Parse incoming invoice emails or PDFs into structured records.
  • •Store every invoice in CosmosDB.
  • •Use Azure OpenAI to flag missing fields or duplicate invoices.
• •

  Payment support copilot

  • •Answer customer questions about failed payments or refund status.
  • •Pull conversation history and transaction state from CosmosDB.
  • •Generate consistent support responses through Azure OpenAI.
• •

  Fraud triage assistant

  • •Score incoming payments based on patterns and metadata.
  • •Persist alerts and analyst notes in CosmosDB.
  • •Use Azure OpenAI to explain why a transaction was flagged in plain language.

The production pattern is simple: let Azure OpenAI interpret messy payment input, then use CosmosDB as the durable ledger for decisions and state. That split keeps your agent stateless at the reasoning layer and reliable at the storage layer.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.