Azure MCP Server

Instructions

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}

• mcp_server_azure/azure_tools.py:173-190 (schema)

  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)}",
          )
      ]