How to Integrate Azure OpenAI for pension funds with CosmosDB for startups

Why this integration matters

If you’re building an AI agent for a pension workflow, Azure OpenAI gives you the reasoning layer: summarization, classification, policy Q&A, and document extraction. CosmosDB gives you the durable state layer: member profiles, claims history, conversation memory, audit trails, and retrieval-friendly JSON storage.

Put them together and you get an agent that can answer questions from pension documents, persist context across sessions, and keep every interaction traceable enough for regulated environments.

Prerequisites

• •An Azure subscription with:
  • •Azure OpenAI resource
  • •Azure Cosmos DB account
• •Deployed Azure OpenAI model:
  • •Chat model deployment name like gpt-4o-mini or your internal deployment alias
• •CosmosDB container created with:
  • •Database name
  • •Container name
  • •Partition key, for example /tenantId
• •Local tooling:
  • •Python 3.10+
  • •pip
• •Python packages:
  • •openai
  • •azure-cosmos
  • •python-dotenv
• •Environment variables set:
  • •AZURE_OPENAI_ENDPOINT
  • •AZURE_OPENAI_API_KEY
  • •AZURE_OPENAI_DEPLOYMENT
  • •COSMOS_ENDPOINT
  • •COSMOS_KEY
  • •COSMOS_DATABASE_NAME
  • •COSMOS_CONTAINER_NAME

Install the SDKs:

pip install openai azure-cosmos python-dotenv

Integration Steps

1. •Set up your configuration and clients.

Keep secrets out of code. Use environment variables and initialize both SDKs once at startup.

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

load_dotenv()

azure_openai_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 = cosmos_client.get_database_client(os.environ["COSMOS_DATABASE_NAME"])
container = database.get_container_client(os.environ["COSMOS_CONTAINER_NAME"])

deployment_name = os.environ["AZURE_OPENAI_DEPLOYMENT"]

2. •Create a helper to read and write agent state in CosmosDB.

For startup systems, store conversation state as JSON documents. Keep the schema simple: tenant, user, session, messages, timestamps.

from datetime import datetime, timezone

def upsert_session_state(tenant_id: str, session_id: str, user_id: str, messages: list):
    doc = {
        "id": session_id,
        "tenantId": tenant_id,
        "userId": user_id,
        "sessionId": session_id,
        "messages": messages,
        "updatedAt": datetime.now(timezone.utc).isoformat(),
        "type": "pension-agent-session"
    }
    container.upsert_item(doc)
    return doc

def get_session_state(session_id: str, tenant_id: str):
    query = """
    SELECT * FROM c
    WHERE c.sessionId = @sessionId AND c.tenantId = @tenantId
    """
    params = [
        {"name": "@sessionId", "value": session_id},
        {"name": "@tenantId", "value": tenant_id},
    ]
    items = list(container.query_items(
        query=query,
        parameters=params,
        enable_cross_partition_query=True
    ))
    return items[0] if items else None

3. •Call Azure OpenAI to process pension-related content.

This is where the agent does work: summarize a pension policy, answer a member question, or extract structured fields from a document.

def ask_azure_openai(question: str, context: str = "") -> str:
    response = azure_openai_client.chat.completions.create(
        model=deployment_name,
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a pension operations assistant. "
                    "Be precise, concise, and flag uncertainty when data is missing."
                ),
            },
            {
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion:\n{question}",
            },
        ],
        temperature=0.2,
        max_tokens=500,
    )
    return response.choices[0].message.content

4. •Wire the two services into one agent flow.

Fetch prior state from CosmosDB, send it to Azure OpenAI, then persist the new turn back to CosmosDB.

def handle_agent_turn(tenant_id: str, session_id: str, user_id: str, user_message: str):
    existing = get_session_state(session_id=session_id, tenant_id=tenant_id)

    prior_messages = existing["messages"] if existing else []
    context_text = "\n".join(
        [f'{m["role"]}: {m["content"]}' for m in prior_messages[-8:]]
    )

    assistant_reply = ask_azure_openai(
        question=user_message,
        context=context_text
    )

    updated_messages = prior_messages + [
        {"role": "user", "content": user_message},
        {"role": "assistant", "content": assistant_reply},
    ]

    upsert_session_state(
        tenant_id=tenant_id,
        session_id=session_id,
        user_id=user_id,
        messages=updated_messages,
    )

    return assistant_reply

5. •Add structured retrieval for startup-grade agent memory.

For real systems, don’t dump everything into one blob forever. Query by tenant or document type when you need targeted context like benefit rules or claim history.

def get_member_documents(tenant_id: str, member_id: str):
    query = """
    SELECT * FROM c
    WHERE c.tenantId = @tenantId AND c.memberId = @memberId AND c.type = @type
    """
    params = [
        {"name": "@tenantId", "value": tenant_id},
        {"name": "@memberId", "value": member_id},
        {"name": "@type", "value": "pension-member-record"},
    ]

    return list(container.query_items(
        query=query,
        parameters=params,
        enable_cross_partition_query=True
    ))

Testing the Integration

Run a simple end-to-end check: send a pension question through Azure OpenAI and confirm the response is stored in CosmosDB.

if __name__ == "__main__":
    tenant_id = "startup-001"
    session_id = "sess-1001"
    user_id = "user-42"

    reply = handle_agent_turn(
        tenant_id=tenant_id,
        session_id=session_id,
        user_id=user_id,
        user_message="Summarize the latest pension contribution status for this member."
    )

    print("Assistant reply:")
    print(reply)

    saved_state = get_session_state(session_id=session_id, tenant_id=tenant_id)
    print("\nSaved messages:", len(saved_state["messages"]))

Expected output:

Assistant reply:
The latest contribution status indicates...
Saved messages: 2

If you want stronger validation in CI/CD:

• •Assert the reply is non-empty.
• •Assert CosmosDB returns the same sessionId.
• •Assert the stored document contains both user and assistant messages.
• •Assert RU usage stays within your budget for query-heavy flows.

Real-World Use Cases

• •Pension member support agents that answer benefit questions from policy docs while storing full conversation history in CosmosDB.
• •Claims triage workflows that classify incoming documents with Azure OpenAI and persist case metadata for downstream processing.
• •Advisor copilots that summarize retirement plans, fetch prior interactions by member ID, and maintain audit-ready state across sessions.

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.