Azure Cosmos Py

MicrosoftOfficialCreated by Microsoft

Skill
Tool integration
Data and AI
MIT
Low risk

From the author

Azure Cosmos DB SDK for Python (NoSQL API). Use for document CRUD, queries, containers, and globally distributed data. Triggers: "cosmos db", "CosmosClient", "container", "document", "NoSQL", "partition key".

— the description by Microsoft

AutoGineer summary

Written by AutoGineer from the skill’s own files, not by the author

The azure-cosmos-py skill provides the Python client library for Azure Cosmos DB NoSQL API. It enables developers to perform CRUD operations, manage databases and containers, execute queries, and handle throughput settings. The skill emphasizes using DefaultAzureCredential for authentication and context managers for resource lifecycle management, while providing guidance on partition key design and query optimization patterns.

What it does

Provides client classes for account, database, and container operations.
Supports document CRUD operations including create, read, replace, upsert, and delete.
Enables SQL-like querying of data with support for parameters and projections.
Facilitates management of container throughput and partition key configuration.
Offers both synchronous and asynchronous client implementations.
Includes patterns for pagination and handling hierarchical partition keys.

When to use it

When developing Python applications that need to interact with Azure Cosmos DB.
When implementing data access layers for NoSQL document storage.
When designing partition strategies for data distribution.
When optimizing query performance through parameterized queries and efficient partition key usage.

Advantages

Provides a unified interface for both sync and async operations.
Integrates with Azure identity services for credential management.
Includes built-in support for handling common database exceptions.
Offers clear patterns for managing resource lifecycles using context managers.

Usage examples

“How do I create a new item in an Azure Cosmos DB container using Python?”: The agent provides a code snippet using container.create_item() with a dictionary representing the item, ensuring the partition key is included.
“How can I query items in Cosmos DB while avoiding cross-partition overhead?”: The agent demonstrates how to use container.query_items() by specifying the partition_key parameter to target a single partition.
“What is the best way to handle authentication in my Cosmos DB Python code?”: The agent explains the use of DefaultAzureCredential and provides a code example showing how to pass it to the CosmosClient.
“How do I implement pagination for a large query result set?”: The agent provides a function using continuation tokens with query_items() to iterate through results page by page.

Install

Download the skill as a ZIP:
https://autogineer.com/agent-skills/microsoft/azure-cosmos-py/download

Unzip it into the cross-client skills folder of your project:

mkdir -p .agents/skills && unzip azure-cosmos-py.zip -d .agents/skills

Documentation by Microsoft

Azure Cosmos DB SDK for Python

Client library for Azure Cosmos DB NoSQL API — globally distributed, multi-model database.

Installation

pip install azure-cosmos azure-identity

Environment Variables

COSMOS_ENDPOINT=https://<account>.documents.azure.com:443/  # Required for all auth methods
COSMOS_DATABASE=mydb  # Required for all auth methods
COSMOS_CONTAINER=mycontainer  # Required for all auth methods
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

Authentication & Lifecycle

🔑 Two rules apply to every code sample below:
Prefer DefaultAzureCredential. It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation.
Local dev: DefaultAzureCredential works as-is.
Production: set AZURE_TOKEN_CREDENTIALS=prod (or AZURE_TOKEN_CREDENTIALS=<specific_credential>) to constrain the credential chain to production-safe credentials.
Wrap every client in a context manager so HTTP transports, sockets, and token caches are released deterministically:
Sync: with <Client>(...) as client:
Async: async with <Client>(...) as client: and async with DefaultAzureCredential() as credential: (from azure.identity.aio)
Snippets may abbreviate this setup, but production code should always follow both rules.

import os
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.cosmos import CosmosClient

# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>
credential = DefaultAzureCredential(require_envvar=True)
# Or use a specific credential directly in production:
# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()

endpoint = "https://<account>.documents.azure.com:443/"

with CosmosClient(url=endpoint, credential=credential) as client:
    # Use client here (see following sections for operations)
    ...

Client Hierarchy

Client	Purpose	Get From
`CosmosClient`	Account-level operations	Direct instantiation
`DatabaseProxy`	Database operations	`client.get_database_client()`
`ContainerProxy`	Container/item operations	`database.get_container_client()`

Core Workflow

Setup Database and Container

# Get or create database
database = client.create_database_if_not_exists(id="mydb")

# Get or create container with partition key
container = database.create_container_if_not_exists(
    id="mycontainer",
    partition_key=PartitionKey(path="/category")
)

# Get existing
database = client.get_database_client("mydb")
container = database.get_container_client("mycontainer")

Create Item

item = {
    "id": "item-001",           # Required: unique within partition
    "category": "electronics",   # Partition key value
    "name": "Laptop",
    "price": 999.99,
    "tags": ["computer", "portable"]
}

created = container.create_item(body=item)
print(f"Created: {created['id']}")

Read Item

# Read requires id AND partition key
item = container.read_item(
    item="item-001",
    partition_key="electronics"
)
print(f"Name: {item['name']}")

Update Item (Replace)

item = container.read_item(item="item-001", partition_key="electronics")
item["price"] = 899.99
item["on_sale"] = True

updated = container.replace_item(item=item["id"], body=item)

Upsert Item

# Create if not exists, replace if exists
item = {
    "id": "item-002",
    "category": "electronics",
    "name": "Tablet",
    "price": 499.99
}

result = container.upsert_item(body=item)

Delete Item

container.delete_item(
    item="item-001",
    partition_key="electronics"
)

Queries

Basic Query

# Query within a partition (efficient)
query = "SELECT * FROM c WHERE c.price < @max_price"
items = container.query_items(
    query=query,
    parameters=[{"name": "@max_price", "value": 500}],
    partition_key="electronics"
)

for item in items:
    print(f"{item['name']}: ${item['price']}")

Cross-Partition Query

# Cross-partition (more expensive, use sparingly)
query = "SELECT * FROM c WHERE c.price < @max_price"
items = container.query_items(
    query=query,
    parameters=[{"name": "@max_price", "value": 500}],
    enable_cross_partition_query=True
)

for item in items:
    print(item)

Query with Projection

query = "SELECT c.id, c.name, c.price FROM c WHERE c.category = @category"
items = container.query_items(
    query=query,
    parameters=[{"name": "@category", "value": "electronics"}],
    partition_key="electronics"
)

Read All Items

# Read all in a partition
items = container.read_all_items()  # Cross-partition
# Or with partition key
items = container.query_items(
    query="SELECT * FROM c",
    partition_key="electronics"
)

Partition Keys

Critical: Always include partition key for efficient operations.

from azure.cosmos import PartitionKey

# Single partition key
container = database.create_container_if_not_exists(
    id="orders",
    partition_key=PartitionKey(path="/customer_id")
)

# Hierarchical partition key (preview)
container = database.create_container_if_not_exists(
    id="events",
    partition_key=PartitionKey(path=["/tenant_id", "/user_id"])
)

Throughput

# Create container with provisioned throughput
container = database.create_container_if_not_exists(
    id="mycontainer",
    partition_key=PartitionKey(path="/pk"),
    offer_throughput=400  # RU/s
)

# Read current throughput
offer = container.read_offer()
print(f"Throughput: {offer.offer_throughput} RU/s")

# Update throughput
container.replace_throughput(throughput=1000)

Async Client

from azure.cosmos.aio import CosmosClient
from azure.identity.aio import DefaultAzureCredential

async def cosmos_operations():
    async with DefaultAzureCredential() as credential:
        async with CosmosClient(endpoint, credential=credential) as client:
            database = client.get_database_client("mydb")
            container = database.get_container_client("mycontainer")
            
            # Create
            await container.create_item(body={"id": "1", "pk": "test"})
            
            # Read
            item = await container.read_item(item="1", partition_key="test")
            
            # Query
            async for item in container.query_items(
                query="SELECT * FROM c",
                partition_key="test"
            ):
                print(item)

import asyncio
asyncio.run(cosmos_operations())

Error Handling

from azure.cosmos.exceptions import CosmosHttpResponseError

try:
    item = container.read_item(item="nonexistent", partition_key="pk")
except CosmosHttpResponseError as e:
    if e.status_code == 404:
        print("Item not found")
    elif e.status_code == 429:
        print(f"Rate limited. Retry after: {e.headers.get('x-ms-retry-after-ms')}ms")
    else:
        raise

Best Practices

Pick sync OR async and stay consistent. Do not mix azure.cosmos sync clients with azure.cosmos.aio async clients in the same call path. Choose one mode per module.
Always use context managers for clients and async credentials. Wrap every client in with CosmosClient(...) as client: (sync) or async with CosmosClient(...) as client: (async). For async DefaultAzureCredential from azure.identity.aio, also use async with credential: so tokens and transports are cleaned up.
Use DefaultAzureCredential for portable auth across local dev and Azure (avoid connection strings / API keys when possible).
Always specify partition key for point reads and queries
Use parameterized queries to prevent injection and improve caching
Avoid cross-partition queries when possible
Use upsert_item for idempotent writes
Use async client for high-throughput scenarios
Design partition key for even data distribution
Use read_item instead of query for single document retrieval

Reference Files

File	Contents
references/partitioning.md	Partition key strategies, hierarchical keys, hot partition detection and mitigation
references/query-patterns.md	Query optimization, aggregations, pagination, transactions, change feed
scripts/setup_cosmos_container.py	CLI tool for creating containers with partitioning, throughput, and indexing

Files

ATTRIBUTION.mdAdded by AutoGineer956 B
LICENSELicence file added by AutoGineer1.1 KiB
SKILL.md9.1 KiB
references/partitioning.md10.3 KiB
references/query-patterns.md10.6 KiB

SKILL.md

Azure Cosmos DB SDK for Python

Client library for Azure Cosmos DB NoSQL API — globally distributed, multi-model database.

Installation

pip install azure-cosmos azure-identity

Environment Variables

COSMOS_ENDPOINT=https://<account>.documents.azure.com:443/  # Required for all auth methods
COSMOS_DATABASE=mydb  # Required for all auth methods
COSMOS_CONTAINER=mycontainer  # Required for all auth methods
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

Authentication & Lifecycle

🔑 Two rules apply to every code sample below:
Prefer DefaultAzureCredential. It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation.
Local dev: DefaultAzureCredential works as-is.
Production: set AZURE_TOKEN_CREDENTIALS=prod (or AZURE_TOKEN_CREDENTIALS=<specific_credential>) to constrain the credential chain to production-safe credentials.
Wrap every client in a context manager so HTTP transports, sockets, and token caches are released deterministically:
Sync: with <Client>(...) as client:
Async: async with <Client>(...) as client: and async with DefaultAzureCredential() as credential: (from azure.identity.aio)
Snippets may abbreviate this setup, but production code should always follow both rules.

import os
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.cosmos import CosmosClient

# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>
credential = DefaultAzureCredential(require_envvar=True)
# Or use a specific credential directly in production:
# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()

endpoint = "https://<account>.documents.azure.com:443/"

with CosmosClient(url=endpoint, credential=credential) as client:
    # Use client here (see following sections for operations)
    ...

Client Hierarchy

Client	Purpose	Get From
`CosmosClient`	Account-level operations	Direct instantiation
`DatabaseProxy`	Database operations	`client.get_database_client()`
`ContainerProxy`	Container/item operations	`database.get_container_client()`

Core Workflow

Setup Database and Container

# Get or create database
database = client.create_database_if_not_exists(id="mydb")

# Get or create container with partition key
container = database.create_container_if_not_exists(
    id="mycontainer",
    partition_key=PartitionKey(path="/category")
)

# Get existing
database = client.get_database_client("mydb")
container = database.get_container_client("mycontainer")

Create Item

item = {
    "id": "item-001",           # Required: unique within partition
    "category": "electronics",   # Partition key value
    "name": "Laptop",
    "price": 999.99,
    "tags": ["computer", "portable"]
}

created = container.create_item(body=item)
print(f"Created: {created['id']}")

Read Item

# Read requires id AND partition key
item = container.read_item(
    item="item-001",
    partition_key="electronics"
)
print(f"Name: {item['name']}")

Update Item (Replace)

item = container.read_item(item="item-001", partition_key="electronics")
item["price"] = 899.99
item["on_sale"] = True

updated = container.replace_item(item=item["id"], body=item)

Upsert Item

# Create if not exists, replace if exists
item = {
    "id": "item-002",
    "category": "electronics",
    "name": "Tablet",
    "price": 499.99
}

result = container.upsert_item(body=item)

Delete Item

container.delete_item(
    item="item-001",
    partition_key="electronics"
)

Queries

Basic Query

# Query within a partition (efficient)
query = "SELECT * FROM c WHERE c.price < @max_price"
items = container.query_items(
    query=query,
    parameters=[{"name": "@max_price", "value": 500}],
    partition_key="electronics"
)

for item in items:
    print(f"{item['name']}: ${item['price']}")

Cross-Partition Query

# Cross-partition (more expensive, use sparingly)
query = "SELECT * FROM c WHERE c.price < @max_price"
items = container.query_items(
    query=query,
    parameters=[{"name": "@max_price", "value": 500}],
    enable_cross_partition_query=True
)

for item in items:
    print(item)

Query with Projection

query = "SELECT c.id, c.name, c.price FROM c WHERE c.category = @category"
items = container.query_items(
    query=query,
    parameters=[{"name": "@category", "value": "electronics"}],
    partition_key="electronics"
)

Read All Items

# Read all in a partition
items = container.read_all_items()  # Cross-partition
# Or with partition key
items = container.query_items(
    query="SELECT * FROM c",
    partition_key="electronics"
)

Partition Keys

Critical: Always include partition key for efficient operations.

from azure.cosmos import PartitionKey

# Single partition key
container = database.create_container_if_not_exists(
    id="orders",
    partition_key=PartitionKey(path="/customer_id")
)

# Hierarchical partition key (preview)
container = database.create_container_if_not_exists(
    id="events",
    partition_key=PartitionKey(path=["/tenant_id", "/user_id"])
)

Throughput

# Create container with provisioned throughput
container = database.create_container_if_not_exists(
    id="mycontainer",
    partition_key=PartitionKey(path="/pk"),
    offer_throughput=400  # RU/s
)

# Read current throughput
offer = container.read_offer()
print(f"Throughput: {offer.offer_throughput} RU/s")

# Update throughput
container.replace_throughput(throughput=1000)

Async Client

from azure.cosmos.aio import CosmosClient
from azure.identity.aio import DefaultAzureCredential

async def cosmos_operations():
    async with DefaultAzureCredential() as credential:
        async with CosmosClient(endpoint, credential=credential) as client:
            database = client.get_database_client("mydb")
            container = database.get_container_client("mycontainer")
            
            # Create
            await container.create_item(body={"id": "1", "pk": "test"})
            
            # Read
            item = await container.read_item(item="1", partition_key="test")
            
            # Query
            async for item in container.query_items(
                query="SELECT * FROM c",
                partition_key="test"
            ):
                print(item)

import asyncio
asyncio.run(cosmos_operations())

Error Handling

from azure.cosmos.exceptions import CosmosHttpResponseError

try:
    item = container.read_item(item="nonexistent", partition_key="pk")
except CosmosHttpResponseError as e:
    if e.status_code == 404:
        print("Item not found")
    elif e.status_code == 429:
        print(f"Rate limited. Retry after: {e.headers.get('x-ms-retry-after-ms')}ms")
    else:
        raise

Best Practices

Pick sync OR async and stay consistent. Do not mix azure.cosmos sync clients with azure.cosmos.aio async clients in the same call path. Choose one mode per module.
Always use context managers for clients and async credentials. Wrap every client in with CosmosClient(...) as client: (sync) or async with CosmosClient(...) as client: (async). For async DefaultAzureCredential from azure.identity.aio, also use async with credential: so tokens and transports are cleaned up.
Use DefaultAzureCredential for portable auth across local dev and Azure (avoid connection strings / API keys when possible).
Always specify partition key for point reads and queries
Use parameterized queries to prevent injection and improve caching
Avoid cross-partition queries when possible
Use upsert_item for idempotent writes
Use async client for high-throughput scenarios
Design partition key for even data distribution
Use read_item instead of query for single document retrieval

Reference Files

File	Contents
references/partitioning.md	Partition key strategies, hierarchical keys, hot partition detection and mitigation
references/query-patterns.md	Query optimization, aggregations, pagination, transactions, change feed
scripts/setup_cosmos_container.py	CLI tool for creating containers with partitioning, throughput, and indexing

Requirements and compatibility

Works with: Claude Code, Claude apps, OpenAI Codex, GitHub Copilot, Cursor, Gemini CLI, Kiro, OpenCode, Goose, Amp, Roo Code, JetBrains Junie, Factory Droid, Devin (ex-Windsurf), Mistral Vibe, OpenHands, Letta

“Works with” means the tool reads the open Agent Skills format; a skill written for one tool may use features others lack.

Needs: Needs an API key, Needs Python

Security

What our scanner noticed· Low risk

Instructions that use the network

A static scan is not a guarantee. Skills are third-party content: review a skill before you let an agent use it.

Attribution and licence

Created by

Microsoft

Publisher

Microsoft (Official)

Source

microsoft/skills · .github/plugins/azure-sdk-python/skills/azure-cosmos-py · commit 0696600

Licence

MIT License (redistribution allowed)

LICENSE (added by AutoGineer from LICENSE)
ATTRIBUTION.md (written by AutoGineer)

Synced

2026-11-09

AutoGineer is not affiliated with or endorsed by Microsoft or Microsoft.

Download ZIP

The ZIP holds the skill’s files, its licence files and an ATTRIBUTION.md. Download azure-cosmos-py.zip

From the author

AutoGineer summary

What it does

When to use it

Advantages

Usage examples

Install

Documentation by Microsoft

Azure Cosmos DB SDK for Python

Installation

Environment Variables

Authentication & Lifecycle

Client Hierarchy

Core Workflow

Setup Database and Container

Create Item

Read Item

Update Item (Replace)

Upsert Item

Delete Item

Queries

Basic Query

Cross-Partition Query

Query with Projection

Read All Items

Partition Keys

Throughput

Async Client

Error Handling

Best Practices

Reference Files

Files

SKILL.md

Azure Cosmos DB SDK for Python

Installation

Environment Variables

Authentication & Lifecycle

Client Hierarchy

Core Workflow

Setup Database and Container

Create Item

Read Item

Update Item (Replace)

Upsert Item

Delete Item

Queries

Basic Query

Cross-Partition Query

Query with Projection

Read All Items

Partition Keys

Throughput

Async Client

Error Handling

Best Practices

Reference Files

Requirements and compatibility

Security

Attribution and licence

Related skills

Anndata

Liteparse

Usfiscaldata

Geopandas

Markitdown

Zarr Python

Download ZIP