blob_upload
Uploads Base64 encoded file content to a specified Blob Storage container on Azure MCP Server, enabling secure and tracked file storage operations.
Instructions
Upload a blob to Blob Storage
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| blob_name | Yes | Name of the blob in the container | |
| container_name | Yes | Name of the Blob Storage container | |
| file_content | Yes | Base64 encoded file content for upload |
Implementation Reference
- mcp_server_azure/azure_server.py:199-205 (handler)Executes the blob_upload tool: retrieves BlobServiceClient, gets blob_client for the container and blob, decodes base64 file_content, and uploads the blob with overwrite=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}
- Defines the input schema and metadata for the blob_upload tool, including 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) with the MCP server by returning the list of Tool objects 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_tools.py:5-112 (registration)Defines and returns the list of Blob Storage tools including the blob_upload Tool, which is used for registration.def get_blob_storage_tools() -> list[Tool]: return [ Tool( name="blob_container_create", description="Create a new Blob Storage container", inputSchema={ "type": "object", "properties": { "container_name": { "type": "string", "description": "Name of the Blob Storage container to create", } }, "required": ["container_name"], }, ), Tool( name="blob_container_list", description="List all Blob Storage containers", inputSchema={"type": "object", "properties": {}}, ), Tool( name="blob_container_delete", description="Delete a Blob Storage container", inputSchema={ "type": "object", "properties": { "container_name": { "type": "string", "description": "Name of the Blob Storage container to delete", } }, "required": ["container_name"], }, ), 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"], }, ), Tool( name="blob_delete", description="Delete a blob from 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 to delete", }, }, "required": ["container_name", "blob_name"], }, ), Tool( name="blob_list", description="List blobs in a Blob Storage container", inputSchema={ "type": "object", "properties": { "container_name": { "type": "string", "description": "Name of the Blob Storage container", } }, "required": ["container_name"], }, ), Tool( name="blob_read", description="Read a blob's content from 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 to read", }, }, "required": ["container_name", "blob_name"], }, ), ]
- mcp_server_azure/azure_server.py:177-238 (handler)The containing async function that dispatches and executes all blob storage tools, including blob_upload, using a shared BlobServiceClient.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)}", ) ]