Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 0 additions & 12 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -84,18 +84,6 @@ make setup
make test
```

## Remote values

Queue and task calls use finite JSON-compatible arguments, return values,
and stream chunks. JSON conversion turns tuples into lists and supported
dictionary keys into strings, including in same-worker remote calls.
Bytes, custom objects, circular references, and non-finite numbers produce
explicit errors. `.local()` remains a direct Python call.

Successful function responses contain `json_result`, including for a null
return value. SDK and worker runtimes share the JSON protocol validators;
use compatible package releases when building worker images.

## ⚡ | Serverless Worker (SDK)

This python package can also be used to create a serverless worker that can be deployed to Runpod as a custom endpoint API.
Expand Down
9 changes: 7 additions & 2 deletions runpod/api/ctl_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from urllib.parse import quote

from runpod import error
from runpod.user_agent import USER_AGENT

from .graphql import run_graphql_query
from .mutations import container_register_auth as container_register_auth_mutations
Expand Down Expand Up @@ -63,9 +64,13 @@ def update_user_settings(pubkey: str, api_key: Optional[str] = None) -> dict:
return get_user(api_key=api_key)


def get_gpus(api_key: Optional[str] = None) -> list[dict]:
def get_gpus(
api_key: Optional[str] = None, *, user_agent: str = USER_AGENT
) -> list[dict]:
"""Get all GPU types."""
response = run_rest_request("GET", "/v2/catalog/gpus", api_key=api_key)
response = run_rest_request(
"GET", "/v2/catalog/gpus", api_key=api_key, user_agent=user_agent
)
return response["gpus"]


Expand Down
12 changes: 11 additions & 1 deletion runpod/api/rest.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,7 @@ def run_rest_request(
timeout: float = 30,
max_retries: int = 3,
max_rate_limit_wait: float = 60,
user_agent: str = USER_AGENT,
) -> Optional[dict[str, Any]]:
"""Send a REST request with bounded 429 retries and shared quota waits.

Expand All @@ -109,6 +110,7 @@ def run_rest_request(
key = _resolve_api_key(api_key)
url = _build_url(path)
headers = _build_headers(key)
headers["User-Agent"] = user_agent
budget = _RetryBudget(max_retries, max_rate_limit_wait, f"{method.upper()} {path}")
with rate_limits.bucket(url, key) as bucket:
for attempt in range(max_retries + 1):
Expand Down Expand Up @@ -167,6 +169,7 @@ async def run_rest_request_async(
max_retries: int = 3,
max_rate_limit_wait: float = 60,
session: Optional[aiohttp.ClientSession] = None,
user_agent: str = USER_AGENT,
) -> Optional[dict[str, Any]]:
"""Send a REST request with cancellable quota waits and bounded 429 retries.

Expand All @@ -178,6 +181,7 @@ async def run_rest_request_async(
key = _resolve_api_key(api_key)
url = _build_url(path)
headers = _build_headers(key)
headers["User-Agent"] = user_agent
budget = _RetryBudget(max_retries, max_rate_limit_wait, f"{method.upper()} {path}")
client_timeout = aiohttp.ClientTimeout(total=timeout)
context = nullcontext(session) if session is not None else aiohttp.ClientSession()
Expand Down Expand Up @@ -233,6 +237,7 @@ async def run_rest_paginated_request_async(
collection: str,
*,
api_key: Optional[str] = None,
user_agent: str = USER_AGENT,
) -> list[dict[str, Any]]:
"""read all cursor pages with one session and four transport attempts per page."""
records = []
Expand All @@ -244,7 +249,12 @@ async def run_rest_paginated_request_async(
for attempt in range(4):
try:
data = await run_rest_request_async(
"GET", path, api_key=api_key, params=params, session=session
"GET",
path,
api_key=api_key,
params=params,
session=session,
user_agent=user_agent,
)
break
except (aiohttp.ClientError, OSError, asyncio.TimeoutError):
Expand Down
44 changes: 36 additions & 8 deletions runpod/apps/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
from ..api.queries import user as user_queries
from ..api.rest import run_rest_paginated_request_async, run_rest_request_async
from ..error import AuthenticationError, QueryError
from ..user_agent import APPS_USER_AGENT

_TRANSPORT_RETRIES = 4

Expand All @@ -31,7 +32,7 @@ async def _execute(self, query: str, task_input: Dict[str, Any]) -> Dict[str, An
query,
variables={"input": task_input},
api_key=(headers["Authorization"][7:] if headers else api_key()),
headers=headers,
headers={**(headers or {}), "User-Agent": APPS_USER_AGENT},
)
return response["data"]

Expand Down Expand Up @@ -228,6 +229,7 @@ async def _execute(
api_key=self._api_key,
variables=variables,
anonymous=anonymous,
headers={"User-Agent": APPS_USER_AGENT},
)
return response["data"]
except (aiohttp.ClientError, OSError, asyncio.TimeoutError):
Expand All @@ -237,6 +239,7 @@ async def _execute(
api_key=self._api_key,
variables=variables,
anonymous=anonymous,
headers={"User-Agent": APPS_USER_AGENT},
)
return response["data"]

Expand All @@ -248,6 +251,7 @@ async def save_endpoint(self, endpoint_input: Dict[str, Any]) -> Dict[str, Any]:
current = await run_rest_request_async(
"GET",
f"/v2/serverless/{_path_segment(endpoint_id)}",
user_agent=APPS_USER_AGENT,
api_key=self._api_key,
)
if not current:
Expand Down Expand Up @@ -281,6 +285,7 @@ async def save_endpoint(self, endpoint_input: Dict[str, Any]) -> Dict[str, Any]:
return await run_rest_request_async(
"PATCH" if endpoint_id else "POST",
path,
user_agent=APPS_USER_AGENT,
api_key=self._api_key,
json=body,
)
Expand All @@ -299,13 +304,14 @@ async def delete_endpoint(self, endpoint_id: str) -> bool:
await run_rest_request_async(
"DELETE",
f"/v2/serverless/{_path_segment(endpoint_id)}",
user_agent=APPS_USER_AGENT,
api_key=self._api_key,
)
return True

async def list_my_endpoints(self) -> List[Dict[str, Any]]:
data = await run_rest_request_async(
"GET", "/v2/serverless", api_key=self._api_key
"GET", "/v2/serverless", user_agent=APPS_USER_AGENT, api_key=self._api_key
)
return data["endpoints"]

Expand All @@ -314,6 +320,7 @@ async def endpoint_workers(self, endpoint_id: str) -> Dict[str, Any]:
return await run_rest_request_async(
"GET",
f"/v2/serverless/{_path_segment(endpoint_id)}/workers",
user_agent=APPS_USER_AGENT,
api_key=self._api_key,
timeout=10,
)
Expand Down Expand Up @@ -352,7 +359,10 @@ async def deploy_task_pod(

async def terminate_pod(self, pod_id: str) -> None:
await run_rest_request_async(
"DELETE", f"/v2/pods/{_path_segment(pod_id)}", api_key=self._api_key
"DELETE",
f"/v2/pods/{_path_segment(pod_id)}",
user_agent=APPS_USER_AGENT,
api_key=self._api_key,
)

async def get_app_by_name(self, name: str) -> Optional[Dict[str, Any]]:
Expand Down Expand Up @@ -388,6 +398,7 @@ async def gpu_stock_status(
data = await run_rest_request_async(
"GET",
f"/v2/catalog/gpus/{_path_segment(gpu_id)}",
user_agent=APPS_USER_AGENT,
api_key=self._api_key,
params={
"include": "AVAILABILITY",
Expand Down Expand Up @@ -424,6 +435,7 @@ async def cpu_stock_status(
data = await run_rest_request_async(
"GET",
f"/v2/catalog/cpus/{_path_segment(cpu['id'])}",
user_agent=APPS_USER_AGENT,
api_key=self._api_key,
params={
"include": "AVAILABILITY",
Expand All @@ -436,7 +448,10 @@ async def cpu_stock_status(
async def network_volume_datacenters(self) -> Set[str]:
"""datacenters supporting any network volume tier chosen by the backend."""
data = await run_rest_request_async(
"GET", "/v2/catalog/datacenters", api_key=self._api_key
"GET",
"/v2/catalog/datacenters",
user_agent=APPS_USER_AGENT,
api_key=self._api_key,
)
datacenters = data.get("dataCenters") if isinstance(data, dict) else None
if not isinstance(datacenters, list):
Expand All @@ -461,14 +476,18 @@ async def network_volume_datacenters(self) -> Set[str]:
async def list_global_volumes(self) -> List[Dict[str, Any]]:
"""list every rest page, retrying transient read failures with backoff."""
return await run_rest_paginated_request_async(
"/v2/global-volumes", "globalVolumes", api_key=self._api_key
"/v2/global-volumes",
"globalVolumes",
user_agent=APPS_USER_AGENT,
api_key=self._api_key,
)

async def create_global_volume(self, name: str) -> Dict[str, Any]:
"""create once via rest and require a usable volume id in the response."""
data = await run_rest_request_async(
"POST",
"/v2/global-volumes",
user_agent=APPS_USER_AGENT,
api_key=self._api_key,
json={"name": name},
)
Expand All @@ -487,7 +506,10 @@ async def create_global_volume(self, name: str) -> Dict[str, Any]:
async def list_network_volumes(self) -> List[Dict[str, Any]]:
"""list every rest page, retrying transient read failures with backoff."""
return await run_rest_paginated_request_async(
"/v2/network-volumes", "networkVolumes", api_key=self._api_key
"/v2/network-volumes",
"networkVolumes",
user_agent=APPS_USER_AGENT,
api_key=self._api_key,
)

async def create_network_volume(
Expand All @@ -496,13 +518,14 @@ async def create_network_volume(
return await run_rest_request_async(
"POST",
"/v2/network-volumes",
user_agent=APPS_USER_AGENT,
api_key=self._api_key,
json={"name": name, "size": size, "dataCenter": data_center_id},
)

async def list_registry_auths(self) -> List[Dict[str, Any]]:
data = await run_rest_request_async(
"GET", "/v2/registries", api_key=self._api_key
"GET", "/v2/registries", user_agent=APPS_USER_AGENT, api_key=self._api_key
)
return data["registries"]

Expand All @@ -512,6 +535,7 @@ async def create_registry_auth(
return await run_rest_request_async(
"POST",
"/v2/registries",
user_agent=APPS_USER_AGENT,
api_key=self._api_key,
json={"name": name, "username": username, "password": password},
)
Expand All @@ -520,6 +544,7 @@ async def delete_registry_auth(self, auth_id: str) -> bool:
await run_rest_request_async(
"DELETE",
f"/v2/registries/{_path_segment(auth_id)}",
user_agent=APPS_USER_AGENT,
api_key=self._api_key,
)
return True
Expand Down Expand Up @@ -603,7 +628,9 @@ async def verify_api_key(self, api_key: str) -> bool:
"""true when api_key authenticates; transport failures still raise."""
try:
response = await run_graphql_query_async(
user_queries.QUERY_USER, api_key=api_key
user_queries.QUERY_USER,
api_key=api_key,
headers={"User-Agent": APPS_USER_AGENT},
)
except (AuthenticationError, QueryError):
return False
Expand Down Expand Up @@ -697,6 +724,7 @@ async def _put_tarball(self, upload_url, reader, total, timeout) -> None:
upload_url,
data=reader,
headers={
"User-Agent": APPS_USER_AGENT,
"Content-Type": "application/gzip",
"Content-Length": str(total),
},
Expand Down
5 changes: 2 additions & 3 deletions runpod/apps/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,15 +17,14 @@

from runpod._hosts import console_url
from runpod.api.rest import _build_url
from runpod.user_agent import USER_AGENT
from runpod.user_agent import APPS_USER_AGENT

import aiohttp

from ..error import QueryError
from .api import AppsApiClient
from .errors import AppError


MAX_KEY_ATTEMPTS = 3

_CLOSED_REASONS = {
Expand Down Expand Up @@ -139,7 +138,7 @@ def load_agent_login(profile: str) -> dict:
def _agent_session() -> aiohttp.ClientSession:
return aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=30),
headers={"Accept": "application/json", "User-Agent": USER_AGENT},
headers={"Accept": "application/json", "User-Agent": APPS_USER_AGENT},
cookie_jar=aiohttp.DummyCookieJar(),
)

Expand Down
12 changes: 10 additions & 2 deletions runpod/apps/gpu.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,8 @@
from enum import Enum
from typing import Any, Dict, List, Optional, Set, Tuple, Union

from ..user_agent import APPS_USER_AGENT

log = logging.getLogger(__name__)


Expand Down Expand Up @@ -286,7 +288,9 @@ def live_pool_members() -> Optional[Dict[str, Set[str]]]:
cache = _catalog_caches.setdefault(context, _CatalogCache())
if cache.may_fetch():
try:
cache.members = _parse_catalog(get_gpus(api_key=context[1]))
cache.members = _parse_catalog(
get_gpus(api_key=context[1], user_agent=APPS_USER_AGENT)
)
except Exception as exc: # noqa: BLE001 - offline, auth, or malformed
cache.fail(exc)
return cache.members
Expand All @@ -303,7 +307,11 @@ async def live_pool_members_async(
if cache.may_fetch():
try:
data = await run_rest_request_async(
"GET", CATALOG_PATH, api_key=context[1], timeout=10
"GET",
CATALOG_PATH,
api_key=context[1],
timeout=10,
user_agent=APPS_USER_AGENT,
)
cache.members = _parse_catalog(data)
except Exception as exc:
Expand Down
3 changes: 2 additions & 1 deletion runpod/apps/logs.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

from ..api.ctl_commands import _path_segment
from ..api.rest import _build_url
from ..user_agent import APPS_USER_AGENT

HAPI_BASE = os.environ.get("RUNPOD_HAPI_URL", "https://hapi.runpod.net")

Expand All @@ -22,7 +23,7 @@
def _headers() -> Dict[str, str]:
from .utils.network import api_key

return {"Authorization": f"Bearer {api_key()}"}
return {"Authorization": f"Bearer {api_key()}", "User-Agent": APPS_USER_AGENT}


async def pod_logs(
Expand Down
4 changes: 2 additions & 2 deletions runpod/apps/targets.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
import aiohttp

from .._hosts import endpoint_domain
from ..user_agent import USER_AGENT
from ..user_agent import APPS_USER_AGENT
from .errors import EndpointNotFound, JobStatusError, RemoteExecutionError
from .protocol import FORMAT_JSON, FunctionRequest
from .serialization import require_json_value
Expand Down Expand Up @@ -43,7 +43,7 @@ def _headers(extra: Optional[Dict[str, str]] = None) -> Dict[str, str]:

headers = {
"Content-Type": "application/json",
"User-Agent": USER_AGENT,
"User-Agent": APPS_USER_AGENT,
}
runtime = runtime_headers()
headers.update(
Expand Down
Loading
Loading