Azure MCP Server

Instructions

Implementation Reference

• mcp_server_azure/azure_server.py:199-205 (handler)

  Executes the blob_upload tool by getting a blob client from BlobServiceClient, base64-decoding the file content, and uploading it to the specified container and blob name with overwrite.

  elif name == "blob_upload":
      blob_client = blob_service_client.get_blob_client(
          container=arguments["container_name"], blob=arguments["blob_name"]
      )
      decoded_content = base64.b64decode(arguments["file_content"])
      blob_client.upload_blob(decoded_content, overwrite=True)
      response = {"blob_name": arguments["blob_name"], "uploaded": True}

• mcp_server_azure/azure_tools.py:40-61 (schema)

  Defines the input schema and Tool metadata for 'blob_upload', specifying required parameters: container_name, blob_name, and base64-encoded file_content.

  Tool(
      name="blob_upload",
      description="Upload a blob to Blob Storage",
      inputSchema={
          "type": "object",
          "properties": {
              "container_name": {
                  "type": "string",
                  "description": "Name of the Blob Storage container",
              },
              "blob_name": {
                  "type": "string",
                  "description": "Name of the blob in the container",
              },
              "file_content": {
                  "type": "string",
                  "description": "Base64 encoded file content for upload",
              },
          },
          "required": ["container_name", "blob_name", "file_content"],
      },
  ),

• mcp_server_azure/azure_server.py:172-176 (registration)

  Registers the blob_upload tool (among others) by returning the list from get_azure_tools() in response to list_tools requests.

  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:177-238 (registration)

  The dispatcher function for all blob_* tools, including blob_upload, which is invoked from the main call_tool handler.

  async def handle_blob_storage_operations(
      azure_rm: AzureResourceManager, name: str, arguments: dict
  ) -> list[TextContent]:
      """Handle Azure Blob Storage operations"""
      blob_service_client = azure_rm.get_blob_service_client()
      response = None
  
      if name == "blob_container_create":
          container_client = blob_service_client.create_container(
              arguments["container_name"]
          )
          response = {
              "container_name": container_client.container_name,
              "created": True,
          }  # Simplify response
      elif name == "blob_container_list":
          containers = blob_service_client.list_containers()
          container_names = [container.name for container in containers]
          response = {"container_names": container_names}
      elif name == "blob_container_delete":
          blob_service_client.delete_container(arguments["container_name"])
          response = {"container_name": arguments["container_name"], "deleted": True}
      elif name == "blob_upload":
          blob_client = blob_service_client.get_blob_client(
              container=arguments["container_name"], blob=arguments["blob_name"]
          )
          decoded_content = base64.b64decode(arguments["file_content"])
          blob_client.upload_blob(decoded_content, overwrite=True)
          response = {"blob_name": arguments["blob_name"], "uploaded": True}
      elif name == "blob_delete":
          blob_client = blob_service_client.get_blob_client(
              container=arguments["container_name"], blob=arguments["blob_name"]
          )
          blob_client.delete_blob()
          response = {"blob_name": arguments["blob_name"], "deleted": True}
      elif name == "blob_list":
          container_client = blob_service_client.get_container_client(
              arguments["container_name"]
          )
          blob_list = container_client.list_blobs()
          blob_names = [blob.name for blob in blob_list]
          response = {"blob_names": blob_names}
      elif name == "blob_read":
          blob_client = blob_service_client.get_blob_client(
              container=arguments["container_name"], blob=arguments["blob_name"]
          )
          downloader = blob_client.download_blob()
          content = downloader.readall().decode("utf-8")
          return [TextContent(type="text", text=content)]
      else:
          raise ValueError(f"Unknown Blob Storage operation: {name}")
  
      azure_rm.log_operation(
          "blob_storage", name.replace("blob_", ""), arguments
      )  # Update service name in log
      return [
          TextContent(
              type="text",
              text=f"Operation Result:\n{json.dumps(response, indent=2, default=custom_json_serializer)}",
          )
      ]

• mcp_server_azure/azure_tools.py:396-401 (registration)

  Combines and returns all Azure tools lists, including blob_upload from get_blob_storage_tools().

  def get_azure_tools() -> list[Tool]:
      return [
          *get_blob_storage_tools(), 
          *get_cosmosdb_tools(),
          *get_app_configuration_tools()
      ]