Python IPFS Upload: Complete Guide with Code Examples

Upload files, images, and JSON to IPFS from Python using the IPFS.NINJA REST API. No SDK required—just the requests library. Complete code examples included.

Nacho Collby Updated: 11 min read
Upload files, images, and JSON to IPFS from Python using the IPFS.NINJA REST API. No SDK required—just the requests library. Complete code examples included.
TL;DR
  • Upload files, images, and JSON to IPFS from Python using nothing but the built-in requests library.
  • No dedicated Python SDK needed — the REST API works the same from any HTTP-capable language.
  • Generate signed upload tokens from Python so browser or mobile clients can upload without your API key.
  • Authenticate every call with a single X-Api-Key header — no OAuth flow or SDK installation required.

To upload files to IPFS from Python in 2026, POST to a pinning service’s REST API using requests.post() with an X-Api-Key header — a permanent CID returns in seconds. The legacy ipfshttpclient package for direct Kubo RPC access is unmaintained since 2022, so the REST-to-pinning-service path is what teams ship. This guide covers JSON, binary files, batch uploads, and signed upload tokens.

Last verified: 2026-08-27.

IPFS Ninja

Set Up Python IPFS Uploads in 5 Minutes#

The fastest way to upload to IPFS from Python is a three-step flow: install requests from PyPI, generate an API key from your pinning service dashboard, then call requests.post() with a JSON body containing your content. Signup to first pinned CID typically takes under two minutes — the requests quickstart docs cover every option you will use below.

  1. Install — add the requests library with pip install requests (no dedicated SDK needed).
  2. Authenticate — set an X-Api-Key header with your IPFS.NINJA API key.
  3. Make your first upload call — send a POST request to /upload/new and get back the CID for your file.

The sections below walk through each step in detail, followed by complete code examples for JSON, file, and batch uploads.

If you’re new to IPFS pinning concepts, check out our comprehensive guide on what IPFS pinning is and how to upload files to IPFS.

IPFS.NINJA advantages for Python developers#

IPFS.NINJA fits Python’s philosophy of readable, dependency-light code. You call one endpoint (POST /upload/new) with the standard requests library — no wheel builds, no C extensions, no Kubo daemon. The service auto-detects JSON, PNG, JPEG, and PDF from the payload, so a single upload function handles every content type your Python app will send.

  • No SDK dependencies: Use Python’s built-in requests library
  • RESTful API: Clean, predictable endpoints that follow HTTP standards
  • JSON-native: Upload JSON data directly without file conversion
  • Multiple authentication methods: API keys, signed tokens, or JWT
  • Built-in analytics: Track upload performance and usage
  • Custom gateways: Branded IPFS access for your applications

Setting Up Your Python Environment#

Any Python 3.11+ interpreter works — the code examples in this guide use only requests, base64, and pathlib from the standard library plus Pillow for image handling later. Install requests from PyPI with pip; a virtualenv is recommended so your IPFS scripts do not collide with system packages. Verify with python --version before continuing.

pip install requests

For image processing examples later in this guide:

pip install Pillow  # For image handling

Authentication Setup#

All IPFS.NINJA API calls require authentication. The simplest method for Python applications is using API keys with the X-Api-Key header.

import requests
import json
import base64
# Your IPFS.NINJA API key (format: bws_ + 32 hex characters)
API_KEY = "bws_1234567890abcdef1234567890abcdef"
BASE_URL = "https://api.ipfs.ninja"
# Default headers for all requests
headers = {
    "X-Api-Key": API_KEY,
    "Content-Type": "application/json"
}

Get your API key from the IPFS.NINJA dashboard after creating your account.

Uploading JSON Data to IPFS#

One of IPFS.NINJA’s unique features is native JSON support. You don’t need to convert JSON to files—upload objects directly:

def upload_json_to_ipfs(data, description=None, metadata=None):
    """
    Upload JSON data directly to IPFS
    Args:
        data (dict): The JSON data to upload
        description (str, optional): Description for the file
        metadata (dict, optional): Additional metadata
    Returns:
        dict: Response containing CID and access URLs
    """
    payload = {
        "content": data,
        "description": description,
        "metadata": metadata
    }
    response = requests.post(
        f"{BASE_URL}/upload/new",
        headers=headers,
        json=payload
    )
    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"Upload failed: {response.status_code} - {response.text}")
# Example: Upload a user profile
user_profile = {
    "name": "Alice Smith",
    "bio": "Python developer and IPFS enthusiast",
    "website": "https://alice.dev",
    "social": {
        "github": "alicesmith",
        "twitter": "@alice_dev"
    }
}
try:
    result = upload_json_to_ipfs(
        data=user_profile,
        description="User profile for Alice Smith",
        metadata={"type": "profile", "version": "1.0"}
    )
    print(f"Upload successful!")
    print(f"CID: {result['cid']}")
    print(f"Size: {result['sizeMB']} MB")
    print(f"IPFS URL: {result['uris']['ipfs']}")
    print(f"Gateway URL: {result['uris']['url']}")
except Exception as e:
    print(f"Upload failed: {e}")

Uploading Files with Base64 Encoding#

For binary files like images, documents, or any file type, use base64 encoding:

def upload_file_to_ipfs(file_path, description=None, metadata=None):
    """
    Upload a file to IPFS using base64 encoding
    Args:
        file_path (str): Path to the file to upload
        description (str, optional): Description for the file
        metadata (dict, optional): Additional metadata
    Returns:
        dict: Response containing CID and access URLs
    """
    try:
        with open(file_path, "rb") as file:
            file_content = base64.b64encode(file.read()).decode('utf-8')
        payload = {
            "content": file_content,
            "description": description or f"File upload: {file_path}",
            "metadata": metadata
        }
        response = requests.post(
            f"{BASE_URL}/upload/new",
            headers=headers,
            json=payload
        )
        if response.status_code == 200:
            return response.json()
        else:
            raise Exception(f"Upload failed: {response.status_code} - {response.text}")
    except FileNotFoundError:
        raise Exception(f"File not found: {file_path}")
# Example: Upload an image
try:
    result = upload_file_to_ipfs(
        file_path="./logo.png",
        description="Company logo",
        metadata={
            "type": "image",
            "format": "png",
            "purpose": "branding"
        }
    )
    print(f"Image uploaded successfully!")
    print(f"CID: {result['cid']}")
    print(f"View at: {result['uris']['url']}")
except Exception as e:
    print(f"Upload failed: {e}")

Batch Upload Function#

For uploading multiple files efficiently:

import os
from pathlib import Path
def batch_upload_files(directory_path, file_extensions=None):
    """
    Upload multiple files from a directory
    Args:
        directory_path (str): Directory containing files to upload
        file_extensions (list, optional): Filter by file extensions (e.g., ['.png', '.jpg'])
    Returns:
        list: Results from each upload
    """
    results = []
    directory = Path(directory_path)
    if not directory.exists():
        raise Exception(f"Directory not found: {directory_path}")
    # Get all files or filter by extensions
    if file_extensions:
        files = []
        for ext in file_extensions:
            files.extend(directory.glob(f"*{ext}"))
    else:
        files = [f for f in directory.iterdir() if f.is_file()]
    print(f"Found {len(files)} files to upload...")
    for file_path in files:
        try:
            print(f"Uploading {file_path.name}...")
            result = upload_file_to_ipfs(
                file_path=str(file_path),
                description=f"Batch upload: {file_path.name}",
                metadata={
                    "batch_upload": True,
                    "original_name": file_path.name,
                    "file_size": file_path.stat().st_size
                }
            )
            results.append({
                "filename": file_path.name,
                "success": True,
                "cid": result['cid'],
                "url": result['uris']['url']
            })
        except Exception as e:
            results.append({
                "filename": file_path.name,
                "success": False,
                "error": str(e)
            })
            print(f"Failed to upload {file_path.name}: {e}")
    return results
# Example usage
try:
    results = batch_upload_files("./images", file_extensions=['.png', '.jpg', '.jpeg'])
    successful = [r for r in results if r['success']]
    failed = [r for r in results if not r['success']]
    print(f"\nSuccessfully uploaded: {len(successful)} files")
    print(f"Failed uploads: {len(failed)} files")
    for result in successful:
        print(f"  - {result['filename']}: {result['url']}")
except Exception as e:
    print(f"Batch upload error: {e}")

Managing Your Pinned Files#

Retrieve and manage your uploaded files:

def list_pinned_files(from_ms=None, to_ms=None):
    """
    List your pinned files
    Args:
        from_ms (int, optional): Start of the time range, Unix ms (defaults to 30 days ago)
        to_ms (int, optional): End of the time range, Unix ms (defaults to now)
    Returns:
        dict: Response containing files list and pagination
    """
    import time
    if to_ms is None:
        to_ms = int(time.time() * 1000)
    if from_ms is None:
        from_ms = to_ms - 30 * 24 * 60 * 60 * 1000
    params = {
        "from": from_ms,
        "to": to_ms
    }
    response = requests.get(
        f"{BASE_URL}/upload/list",
        headers=headers,
        params=params
    )
    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"Failed to list files: {response.status_code} - {response.text}")
def get_file_details(cid):
    """
    Get details for a specific file
    Args:
        cid (str): The Content Identifier of the file
    Returns:
        dict: File details
    """
    response = requests.get(
        f"{BASE_URL}/file/{cid}",
        headers=headers
    )
    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"Failed to get file details: {response.status_code} - {response.text}")
# Example: List recent uploads
try:
    files = list_pinned_files()
    print(f"Total files: {files.get('total', 0)}")
    print("\nRecent uploads:")
    for file_info in files.get('files', []):
        print(f"  - {file_info['cid'][:12]}... | {file_info['sizeMB']} MB | {file_info['description']}")
except Exception as e:
    print(f"Failed to list files: {e}")

Pin Existing IPFS Content#

Pin content that already exists on IPFS:

def pin_existing_cid(cid, description=None):
    """
    Pin existing IPFS content by CID
    Args:
        cid (str): The Content Identifier to pin
        description (str, optional): Description for the pinned content
    Returns:
        dict: Response with pin status
    """
    payload = {
        "cid": cid,
        "description": description or f"Pinned existing content: {cid}"
    }
    response = requests.post(
        f"{BASE_URL}/pin",
        headers=headers,
        json=payload
    )
    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"Pin failed: {response.status_code} - {response.text}")
# Example: Pin popular NFT metadata
try:
    result = pin_existing_cid(
        cid="QmYwAPJzv5CZsnA625s3Xf2nemtYgPpHdWEz79ojWnPbdG",
        description="Popular NFT metadata backup"
    )
    print(f"Successfully pinned existing content!")
    print(f"CID: {result['cid']}")
except Exception as e:
    print(f"Pin failed: {e}")

Using Signed Upload Tokens#

For secure client-side uploads or temporary access, use signed tokens:

def create_upload_token(expires_in_hours=24, max_uses=None):
    """
    Create a signed upload token for temporary uploads
    Args:
        expires_in_hours (int): Token expiration time in hours
        max_uses (int, optional): Maximum number of uses (unlimited if None)
    Returns:
        dict: Token details including the signed token string
    """
    from datetime import datetime, timedelta, timezone
    expiry_time = datetime.now(timezone.utc) + timedelta(hours=expires_in_hours)
    payload = {
        "expiry": expiry_time.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "maxUses": max_uses
    }
    response = requests.post(
        f"{BASE_URL}/upload/signed-url",
        headers=headers,
        json=payload
    )
    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"Token creation failed: {response.status_code} - {response.text}")
def upload_with_token(data, token, description=None):
    """
    Upload using a signed token instead of API key
    Args:
        data: Content to upload (dict for JSON, base64 string for files)
        token (str): Signed upload token
        description (str, optional): Description for the upload
    Returns:
        dict: Upload response
    """
    token_headers = {
        "Authorization": f"Signed {token}",
        "Content-Type": "application/json"
    }
    payload = {
        "content": data,
        "description": description
    }
    response = requests.post(
        f"{BASE_URL}/upload/new",
        headers=token_headers,
        json=payload
    )
    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"Token upload failed: {response.status_code} - {response.text}")
# Example: Create token for client-side uploads
try:
    token_info = create_upload_token(expires_in_hours=1, max_uses=10)
    upload_token = token_info['token']
    print(f"Created upload token: {upload_token[:20]}...")
    print(f"Expires: {token_info['expiry']}")
    print(f"Max uses: {token_info.get('maxUses', 'unlimited')}")
    # Use the token for an upload
    test_data = {"message": "Hello from signed token!"}
    result = upload_with_token(test_data, upload_token, "Token test upload")
    print(f"Token upload successful: {result['cid']}")
except Exception as e:
    print(f"Token operation failed: {e}")

Complete Example: NFT Metadata Uploader#

Here’s a practical example that combines multiple concepts for NFT metadata management:

class NFTMetadataUploader:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.ipfs.ninja"
        self.headers = {
            "X-Api-Key": api_key,
            "Content-Type": "application/json"
        }
    def upload_nft_image(self, image_path, name, description=""):
        """Upload NFT image and return IPFS URL"""
        try:
            with open(image_path, "rb") as file:
                image_content = base64.b64encode(file.read()).decode('utf-8')
            payload = {
                "content": image_content,
                "description": f"NFT Image: {name}",
                "metadata": {
                    "type": "nft_image",
                    "name": name,
                    "description": description
                }
            }
            response = requests.post(
                f"{self.base_url}/upload/new",
                headers=self.headers,
                json=payload
            )
            if response.status_code == 200:
                result = response.json()
                return result['uris']['ipfs']  # Return IPFS URI
            else:
                raise Exception(f"Image upload failed: {response.text}")
        except Exception as e:
            raise Exception(f"Failed to upload image {image_path}: {e}")
    def upload_nft_metadata(self, name, description, image_ipfs_url, attributes=None, external_url=None):
        """Upload complete NFT metadata"""
        metadata = {
            "name": name,
            "description": description,
            "image": image_ipfs_url,
        }
        if attributes:
            metadata["attributes"] = attributes
        if external_url:
            metadata["external_url"] = external_url
        payload = {
            "content": metadata,
            "description": f"NFT Metadata: {name}",
            "metadata": {
                "type": "nft_metadata",
                "standard": "ERC-721"
            }
        }
        response = requests.post(
            f"{self.base_url}/upload/new",
            headers=self.headers,
            json=payload
        )
        if response.status_code == 200:
            return response.json()
        else:
            raise Exception(f"Metadata upload failed: {response.text}")
    def create_complete_nft(self, image_path, name, description, attributes=None, external_url=None):
        """Upload image and metadata for a complete NFT"""
        print(f"Creating NFT: {name}")
        # Step 1: Upload image
        print("  1. Uploading image...")
        image_ipfs_url = self.upload_nft_image(image_path, name, description)
        print(f"     Image IPFS URL: {image_ipfs_url}")
        # Step 2: Upload metadata
        print("  2. Uploading metadata...")
        metadata_result = self.upload_nft_metadata(
            name=name,
            description=description,
            image_ipfs_url=image_ipfs_url,
            attributes=attributes,
            external_url=external_url
        )
        print(f"     Metadata CID: {metadata_result['cid']}")
        print(f"     Metadata URL: {metadata_result['uris']['url']}")
        return {
            "image_ipfs": image_ipfs_url,
            "metadata_cid": metadata_result['cid'],
            "metadata_url": metadata_result['uris']['url'],
            "metadata_ipfs": metadata_result['uris']['ipfs']
        }
# Example usage
if __name__ == "__main__":
    uploader = NFTMetadataUploader("bws_1234567890abcdef1234567890abcdef")
    try:
        nft_result = uploader.create_complete_nft(
            image_path="./nft_artwork.png",
            name="Cosmic Cat #001",
            description="A rare cosmic cat exploring the digital universe",
            attributes=[
                {"trait_type": "Background", "value": "Space"},
                {"trait_type": "Species", "value": "Cosmic Cat"},
                {"trait_type": "Rarity", "value": "Legendary"},
                {"trait_type": "Power Level", "value": 95}
            ],
            external_url="https://myproject.com/nft/1"
        )
        print("\n🎉 NFT created successfully!")
        print(f"Use this metadata URI in your smart contract: {nft_result['metadata_ipfs']}")
    except Exception as e:
        print(f"NFT creation failed: {e}")

Error Handling and Best Practices#

Implement robust error handling for production applications:

import time
from functools import wraps
def retry_on_failure(max_retries=3, delay=1):
    """Decorator to retry failed uploads"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise e
                    print(f"Attempt {attempt + 1} failed: {e}")
                    print(f"Retrying in {delay} seconds...")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator
@retry_on_failure(max_retries=3, delay=2)
def robust_upload(file_path, description=None):
    """Upload with automatic retry on failure"""
    return upload_file_to_ipfs(file_path, description)
def validate_api_response(response, operation="operation"):
    """Validate API response and provide detailed error info"""
    if response.status_code == 200:
        return response.json()
    elif response.status_code == 401:
        raise Exception(f"Authentication failed for {operation}. Check your API key.")
    elif response.status_code == 413:
        raise Exception(f"File too large for {operation}. Check your plan limits.")
    elif response.status_code == 429:
        raise Exception(f"Rate limit exceeded for {operation}. Please wait and retry.")
    else:
        raise Exception(f"{operation} failed: {response.status_code} - {response.text}")

Next Steps#

You now have a complete toolkit for uploading files to IPFS from Python! This covers:

  • JSON and file uploads with the IPFS.NINJA API
  • Batch processing for multiple files
  • File management and pinning existing content
  • Secure uploads with signed tokens
  • Production-ready error handling

For more advanced features, explore our other guides:

The IPFS.NINJA API provides everything you need for Python-based IPFS applications without the complexity of running your own infrastructure.

Ready to start pinning? Start your 7-day trial — 10 GB / 200 files / 20 GB bandwidth / 1 dedicated gateway (trial file cap — Bodhi lifts to unlimited). No credit card required. Then Bodhi $5/mo, Karma $19/mo, or Nirvana $59/mo.

Nacho Coll

About the author

Founder & Engineer at IPFS.NINJA

Nacho founded IPFS.NINJA to make content-addressed storage feel as simple as an S3 PUT — a single API call, a permanent CID, no wallets or peer discovery to reason about. Writes about IPFS internals, decentralized storage patterns, and the pinning-service landscape from the operator side of the wire.

Back to Blog

Related Posts

Go IPFS Upload: Pin Files from Golang Applications
Tutorials

Go IPFS Upload: Pin Files from Golang Applications

Jul 20, 20268min read
Connect Claude Code to IPFS with our MCP server (5-minute setup)
Tutorials

Connect Claude Code to IPFS with our MCP server (5-minute setup)

Jul 11, 20266min read
IPFS + React: Build a Decentralized File Upload Component
Tutorials

IPFS + React: Build a Decentralized File Upload Component

May 11, 202612min read