Azure Cosmos DB

Install the package

dotnet add package Microsoft.Agents.AI.CosmosNoSql --prerelease

Configure Cosmos DB chat history

Use the managed-identity extension to attach CosmosChatHistoryProvider to ChatClientAgentOptions.

using Azure.Identity;
using Microsoft.Agents.AI;

var options = new ChatClientAgentOptions
{
    ChatOptions = new() { Instructions = "You are a helpful assistant." }
}.WithCosmosDBChatHistoryProviderUsingManagedIdentity(
    accountEndpoint: Environment.GetEnvironmentVariable("AZURE_COSMOS_ENDPOINT")!,
    databaseId: Environment.GetEnvironmentVariable("AZURE_COSMOS_DATABASE_NAME")!,
    containerId: Environment.GetEnvironmentVariable("AZURE_COSMOS_CONTAINER_NAME")!,
    tokenCredential: new DefaultAzureCredential());

AIAgent agent = chatClient.AsAIAgent(options);

The default state initializer creates a conversation ID. Supply a CosmosChatHistoryProvider.State initializer when your application needs explicit conversation, tenant, and user routing. When tenant and user IDs are present, the provider uses a hierarchical partition key.

Install the package

pip install agent-framework-azure-cosmos --pre

Configure CosmosHistoryProvider

The Python provider accepts either an Azure credential or an account key. It uses session_id as the partition key. Provider instances that use the same account, database, container, nonempty session_id, and source_id access the same persisted history. The source_id filters history within the partition. These identifiers select stored history; they aren't authentication or authorization boundaries. Bind them to authenticated and authorized application context, and use distinct trusted namespaces when you need isolation.

# 1. Create an Azure credential and a CosmosHistoryProvider for agent context
async with (
    AzureCliCredential() as credential,
    CosmosHistoryProvider(
        endpoint=cosmos_endpoint,
        database_name=cosmos_database_name,
        container_name=cosmos_container_name,
        credential=cosmos_key or credential,
    ) as history_provider,
    # 2. Create an agent that uses Cosmos for persisted conversation history.
    Agent(
        client=FoundryChatClient(
            project_endpoint=project_endpoint,
            model=model,
            credential=credential,
        ),
        name="CosmosHistoryAgent",
        instructions="You are a helpful assistant that remembers prior turns.",
        context_providers=[history_provider],
        default_options={"store": False},
    ) as agent,
):
    # 3. Create a session (session_id is used as the partition key).
    session = agent.create_session()

    # 4. Run a multi-turn conversation; history is persisted by CosmosHistoryProvider.
    response1 = await agent.run("My name is Ada and I enjoy distributed systems.", session=session)
    print(f"Assistant: {response1.text}")

    response2 = await agent.run("What do you remember about me?", session=session)
    print(f"Assistant: {response2.text}")
    print(f"Container: {history_provider.container_name}")

Persist the serialized AgentSession in trusted application storage when clients need to recover the same session identifier later.

Prerequisites

• An Azure Cosmos DB account and database.
• A Microsoft Foundry project with chat and embedding model deployments.
• Azure identity access to both resources.

Install the packages

pip install agent-framework-azure-cosmos-memory agent-framework-foundry --pre

Configure the memory provider

The same Foundry project can supply the chat model, embeddings, and memory extraction model. Attach the provider through context_providers.

def _foundry_project_endpoint() -> str:
    """Read the preferred project endpoint variable with legacy compatibility."""
    return os.environ.get("FOUNDRY_PROJECT_ENDPOINT") or os.environ["FOUNDRY_ENDPOINT"]


def _build_agent(provider: CosmosMemoryContextProvider, credential: DefaultAzureCredential) -> Agent:
    """Build an agent that uses the memory provider and the same Foundry endpoint for chat."""
    return Agent(
        client=FoundryChatClient(
            project_endpoint=_foundry_project_endpoint(),
            model=os.getenv("CHAT_MODEL", "gpt-4o-mini"),
            credential=credential,
        ),
        name="Memory Assistant",
        instructions="You are a helpful assistant with long-term memory about the user.",
        context_providers=[provider],
    )


async def user_scoped_memory() -> None:
    """Memory scoped to a stable user id, so it persists across sessions and threads."""
    credential = DefaultAzureCredential()
    provider = CosmosMemoryContextProvider(
        cosmos_endpoint=os.environ["COSMOS_ENDPOINT"],
        foundry_endpoint=_foundry_project_endpoint(),
        embedding_model=os.getenv("EMBEDDING_MODEL", "text-embedding-3-large"),
        chat_model=os.getenv("CHAT_MODEL", "gpt-4o-mini"),
        credential=credential,
    )
    agent = _build_agent(provider, credential)

    async with provider:
        session = agent.create_session()
        # Provider state is scoped by source id; set a stable user id there so memory
        # persists across sessions rather than being limited to this one.
        session.state.setdefault(provider.source_id, {})["user_id"] = "alice"
        first = await agent.run("I love hiking and I'm allergic to peanuts.", session=session)
        print("Assistant:", first.text)

        # A brand-new session for the same user still recalls the earlier facts.
        new_session = agent.create_session()

A stable user_id keeps memory available across sessions and threads. Store it under the provider-owned state namespace: session.state[provider.source_id]["user_id"]. Don't put it in flat session state, where another provider or application field can collide with it. Without a user ID, the provider scopes memory to the current session ID.

Memory processing

Memory extraction runs in the background after each turn. Use the provider as an async context manager or call flush() before shutdown so pending extraction completes before the clients close.

The provider also supports custom extraction prompts, processor cadence, confidence thresholds, memory types, and retrieval limits. Select facts, procedures, and episodes with memory_types. With Agent Memory Toolkit 0.3.0b2 or later, facts and selected episodes share the ranked top_k result limit. Selected procedures are compiled separately for the current task and don't consume that limit. Older supported Toolkit versions retain their generic retrieval behavior. Set top_k to a positive integer and min_confidence to a value from 0.0 through 1.0, inclusive. Invalid values raise ValueError when you create the provider.

Go provides a Cosmos DB for NoSQL vector store through azureprovider.NewCosmosDB. It uses an existing azcosmos.Client and database, with collections exposed through vectordata.

The connector supports generated embeddings, vector search, and full-text/vector hybrid search. The account must have the required search capabilities enabled. Ranked queries require equality filters for every configured partition-key property; cross-partition ranking isn't supported by this connector.

The Cosmos DB Go sample requires AZURE_COSMOS_ENDPOINT, AZURE_COSMOS_DATABASE, FOUNDRY_PROJECT_ENDPOINT, and FOUNDRY_EMBEDDING_MODEL. It uses a 3,072-dimensional schema and creates and deletes a uniquely named temporary container in the existing database.

This connector isn't the dedicated long-term memory provider shown in the C# example. For application-managed retrieval, compose collection searches with text-search RAG, preserving the connector's partition filters and your application's access controls. See Vector store integrations for the available Go stores.