cosmosdb_container_delete
Delete a Cosmos DB container to remove unnecessary data storage and manage database resources effectively. Specify the container name and optional database name to execute the deletion.
Instructions
Delete a Cosmos DB container
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| container_name | Yes | Name of the Cosmos DB container | |
| database_name | No | Name of the Cosmos DB database (optional, defaults to 'defaultdb') |
Implementation Reference
- mcp_server_azure/azure_server.py:263-265 (handler)Executes the deletion of the Cosmos DB container by calling database.delete_container with the provided container name.elif name == "cosmosdb_container_delete": # Renamed from table to container database.delete_container(arguments["container_name"]) response = {"container_name": arguments["container_name"], "deleted": True}
- Defines the Tool schema including name, description, and inputSchema for validating arguments (container_name required, database_name optional).Tool( name="cosmosdb_container_delete", # Renamed from table to container description="Delete a Cosmos DB container", # Updated description inputSchema={ "type": "object", "properties": { "container_name": { # Renamed from table_name "type": "string", "description": "Name of the Cosmos DB container", # Updated description }, "database_name": { "type": "string", "description": "Name of the Cosmos DB database (optional, defaults to 'defaultdb')", }, }, "required": ["container_name"], }, ),
- mcp_server_azure/azure_server.py:171-175 (registration)Registers the cosmosdb_container_delete tool (among others) by returning get_azure_tools() in the list_tools handler.@server.list_tools() async def list_tools() -> list[Tool]: """List available Azure tools""" logger.debug("Handling list_tools request") return get_azure_tools() # Use get_azure_tools
- mcp_server_azure/azure_server.py:436-439 (handler)Dispatches cosmosdb_container_delete calls to the specific handle_cosmosdb_operations function based on name prefix.elif name.startswith("cosmosdb_"): # Updated prefix to cosmosdb_ return await handle_cosmosdb_operations( azure_rm, name, arguments ) # Use cosmosdb handler
- mcp_server_azure/azure_server.py:239-327 (handler)Comprehensive handler function for all cosmosdb tools, including setup of cosmos_client and database, dispatch via if-elif chain, logging, and response formatting. The specific delete logic is included.async def handle_cosmosdb_operations( azure_rm: AzureResourceManager, name: str, arguments: dict ) -> list[TextContent]: """Handle Azure Cosmos DB operations (NoSQL API)""" cosmos_client = azure_rm.get_cosmos_client() logger.log(cosmos_client) database = cosmos_client.get_database_client( arguments.get("database_name", "SampleDB") ) # Assuming a default db response = None if name == "cosmosdb_container_create": # Renamed from table to container container = database.create_container( id=arguments["container_name"], partition_key=arguments["partition_key"] ) response = {"container_id": container.id, "created": True} elif name == "cosmosdb_container_describe": # Renamed from table to container container = database.get_container_client(arguments["container_name"]) container_properties = container.read() response = container_properties elif name == "cosmosdb_container_list": # Renamed from table to container containers = list(database.list_containers()) container_names = [c["id"] for c in containers] response = {"container_names": container_names} elif name == "cosmosdb_container_delete": # Renamed from table to container database.delete_container(arguments["container_name"]) response = {"container_name": arguments["container_name"], "deleted": True} elif ( name == "cosmosdb_item_create" ): # Renamed from put to create, and table to container container_client = database.get_container_client( arguments["container_name"] ) item = container_client.create_item(body=arguments["item"]) response = {"item_id": item["id"], "created": True} elif ( name == "cosmosdb_item_read" ): # Renamed from get to read, and table to container container_client = database.get_container_client( arguments["container_name"] ) item = container_client.read_item( item=arguments["item_id"], partition_key=arguments["partition_key"] ) response = item elif ( name == "cosmosdb_item_replace" ): # Renamed from update to replace, and table to container, using replace_item for full replace container_client = database.get_container_client( arguments["container_name"] ) item = container_client.replace_item( item=arguments["item_id"], body=arguments["item"] ) response = {"item_id": item["id"], "replaced": True} elif name == "cosmosdb_item_delete": # Renamed table to container container_client = database.get_container_client( arguments["container_name"] ) container_client.delete_item( item=arguments["item_id"], partition_key=arguments["partition_key"] ) response = {"item_id": arguments["item_id"], "deleted": True} elif ( name == "cosmosdb_item_query" ): # Renamed table to container, simplified query container_client = database.get_container_client( arguments["container_name"] ) items = list( container_client.query_items( query=arguments["query"], parameters=arguments.get("parameters", []), # Optional parameters ) ) response = {"items": items} else: raise ValueError(f"Unknown Cosmos DB operation: {name}") azure_rm.log_operation( "cosmosdb", name.replace("cosmosdb_", ""), arguments ) # Update service name in log return [ TextContent( type="text", text=f"Operation Result:\n{json.dumps(response, indent=2, default=custom_json_serializer)}", ) ]