Welcome to the official Python library for Runpod API & SDK.
- Table of Contents
- π» | Installation
- β‘ | Serverless Worker (SDK)
- π | API Language Library (GraphQL Wrapper)
- π | Directory
- π€ | Community and Contributing
# Install with pip
pip install runpod
# Install with uv (faster alternative)
uv add runpod
To get the latest changes that haven't been released to PyPI yet:
# Install latest development version from main branch with pip
pip install git+https://github.com/runpod/runpod-python.git
# Install with uv
uv add git+https://github.com/runpod/runpod-python.git
# Install a specific branch
pip install git+https://github.com/runpod/runpod-python.git@branch-name
# Install a specific tag/release
pip install git+https://github.com/runpod/runpod-python.git@v1.0.0
# Install in editable mode for development
git clone https://github.com/runpod/runpod-python.git
cd runpod-python
pip install -e .
Python 3.8 or higher is required to use the latest version of this package.
This python package can also be used to create a serverless worker that can be deployed to Runpod as a custom endpoint API.
Create a python script in your project that contains your model definition and the Runpod worker start code. Run this python code as your default container start command:
# my_worker.py
import runpod
def is_even(job):
job_input = job["input"]
the_number = job_input["number"]
if not isinstance(the_number, int):
return {"error": "Silly human, you need to pass an integer."}
if the_number % 2 == 0:
return True
return False
runpod.serverless.start({"handler": is_even})
Make sure that this file is ran when your container starts. This can be accomplished by calling it in the docker command when you set up a template at console.runpod.io/serverless/user/templates or by setting it as the default command in your Dockerfile.
See our blog post for creating a basic Serverless API, or view the details docs for more information.
You can also test your worker locally before deploying it to Runpod. This is useful for debugging and testing.
python my_worker.py --rp_serve_api
When interacting with the Runpod API you can use this library to make requests to the API.
import runpod
runpod.api_key = "your_runpod_api_key_found_under_settings"
You can interact with Runpod endpoints via a run
or run_sync
method.
endpoint = runpod.Endpoint("ENDPOINT_ID")
run_request = endpoint.run(
{"your_model_input_key": "your_model_input_value"}
)
# Check the status of the endpoint run request
print(run_request.status())
# Get the output of the endpoint run request, blocking until the endpoint run is complete.
print(run_request.output())
endpoint = runpod.Endpoint("ENDPOINT_ID")
run_request = endpoint.run_sync(
{"your_model_input_key": "your_model_input_value"}
)
# Returns the job results if completed within 90 seconds, otherwise, returns the job status.
print(run_request )
The SDK supports multiple ways to set API keys:
1. Global API Key (Default)
import runpod
# Set global API key
runpod.api_key = "your_runpod_api_key"
# All endpoints will use this key by default
endpoint = runpod.Endpoint("ENDPOINT_ID")
result = endpoint.run_sync({"input": "data"})
2. Endpoint-Specific API Key
# Create endpoint with its own API key
endpoint = runpod.Endpoint("ENDPOINT_ID", api_key="specific_api_key")
# This endpoint will always use the provided API key
result = endpoint.run_sync({"input": "data"})
The SDK uses this precedence order (highest to lowest):
- Endpoint instance API key (if provided to
Endpoint()
) - Global API key (set via
runpod.api_key
)
import runpod
# Example showing precedence
runpod.api_key = "GLOBAL_KEY"
# This endpoint uses GLOBAL_KEY
endpoint1 = runpod.Endpoint("ENDPOINT_ID")
# This endpoint uses ENDPOINT_KEY (overrides global)
endpoint2 = runpod.Endpoint("ENDPOINT_ID", api_key="ENDPOINT_KEY")
# All requests from endpoint2 will use ENDPOINT_KEY
result = endpoint2.run_sync({"input": "data"})
Each Endpoint
instance maintains its own API key, making concurrent operations safe:
import threading
import runpod
def process_request(api_key, endpoint_id, input_data):
# Each thread gets its own Endpoint instance
endpoint = runpod.Endpoint(endpoint_id, api_key=api_key)
return endpoint.run_sync(input_data)
# Safe concurrent usage with different API keys
threads = []
for customer in customers:
t = threading.Thread(
target=process_request,
args=(customer["api_key"], customer["endpoint_id"], customer["input"])
)
threads.append(t)
t.start()
import runpod
runpod.api_key = "your_runpod_api_key_found_under_settings"
# Get all my pods
pods = runpod.get_pods()
# Get a specific pod
pod = runpod.get_pod(pod.id)
# Create a pod with GPU
pod = runpod.create_pod("test", "runpod/stack", "NVIDIA GeForce RTX 3070")
# Create a pod with CPU
pod = runpod.create_pod("test", "runpod/stack", instance_id="cpu3c-2-4")
# Stop the pod
runpod.stop_pod(pod.id)
# Resume the pod
runpod.resume_pod(pod.id)
# Terminate the pod
runpod.terminate_pod(pod.id)
.
βββ docs # Documentation
βββ examples # Examples
βββ runpod # Package source code
β βββ api_wrapper # Language library - API (GraphQL)
β βββ cli # Command Line Interface Functions
β βββ endpoint # Language library - Endpoints
β βββ serverless # SDK - Serverless Worker
βββ tests # Package tests
We welcome both pull requests and issues on GitHub. Bug fixes and new features are encouraged, but please read our contributing guide first.