The official Python SDK for OpenFaaS.
- Manage functions, namespaces, and secrets
- Invoke functions synchronously or asynchronously
- Stream function logs
- Basic auth and OpenFaaS IAM support
- Build and push function images from source via the Function Builder API
pip install git+https://github.com/openfaas/python-sdk.gitfrom openfaas_sdk import Client, BasicAuth
client = Client(
gateway_url="https://gateway.example.com",
auth=BasicAuth("admin", "secret"),
)
functions = client.get_functions("openfaas-fn")
for fn in functions:
print(fn.name, fn.replicas)
client.close()Use the client as a context manager to ensure connections are closed:
from openfaas_sdk import Client, BasicAuth
with Client("https://gateway.example.com", auth=BasicAuth("admin", "secret")) as client:
functions = client.get_functions("openfaas-fn")from openfaas_sdk import BasicAuth
auth = BasicAuth(username="admin", password="secret")The password can be read from a file:
from openfaas_sdk import BasicAuth
with open("/var/secrets/basic-auth-password") as f:
password = f.read().strip()
auth = BasicAuth(username="admin", password=password)For workloads outside Kubernetes, use ClientCredentialsTokenSource to obtain tokens from an external IdP and exchange them for an OpenFaaS gateway JWT:
from openfaas_sdk import Client, TokenAuth, ClientCredentialsTokenSource
ts = ClientCredentialsTokenSource(
client_id="my-app",
client_secret="secret",
token_url="https://idp.example.com/realms/master/protocol/openid-connect/token",
scope="openid",
)
auth = TokenAuth(
token_url="https://gateway.example.com/oauth/token",
token_source=ts,
)
with Client("https://gateway.example.com", auth=auth) as client:
functions = client.get_functions("openfaas-fn")When running inside a Kubernetes cluster with OpenFaaS IAM enabled, use TokenAuth with ServiceAccountTokenSource to exchange the pod's projected service account token for an OpenFaaS gateway JWT automatically:
from openfaas_sdk import Client, TokenAuth, ServiceAccountTokenSource
auth = TokenAuth(
token_url="https://gateway.example.com/oauth/token",
token_source=ServiceAccountTokenSource(),
)
with Client("https://gateway.example.com", auth=auth) as client:
functions = client.get_functions("openfaas-fn")ServiceAccountTokenSource re-reads /var/secrets/tokens/openfaas-token on every call so that Kubernetes token rotation is handled transparently. The path can be overridden with the token_mount_path environment variable.
TokenAuth caches the exchanged gateway token and refreshes it automatically when it expires (10-second expiry buffer).
TokenAuth also implements the TokenSource protocol, so it is automatically used as the function_token_source for per-function scoped token exchange when calling get_function_token().
If the built-in token sources don't fit your setup, you can implement your own. The TokenSource protocol requires a single method, token() -> str, that returns a raw OIDC JWT. The implementation can contain any logic you need, for example reading a token from a file, calling an external API, or signing a JWT locally:
class FileTokenSource:
"""Read a token from a file on each call."""
def __init__(self, path: str) -> None:
self._path = path
def token(self) -> str:
with open(self._path) as f:
return f.read().strip()Wire it up with TokenAuth the same way as any other token source:
ts = FileTokenSource("/var/run/secrets/oidc-token")
auth = TokenAuth(
token_url="https://gateway.example.com/oauth/token",
token_source=ts,
)Advanced use case. For most scenarios, use
invoke_functionwithuse_function_auth=True, which handles token exchange automatically. The method below is for cases where you need full control over function invocation outside of the SDK.
get_function_token() exchanges the current identity token for a short-lived token scoped to a specific function (audience "<namespace>:<function-name>"). Use this token when invoking the function directly without going through the SDK:
token = client.get_function_token("my-func", "openfaas-fn")
# token is a raw JWT string — pass it as a Bearer token when invoking the functioninfo = client.get_info()
# info.arch, info.provider.orchestration, info.version.release# List all functions in a namespace
functions = client.get_functions("openfaas-fn")
# Get a single function
fn = client.get_function("env", "openfaas-fn")
# fn.name, fn.replicas, fn.available_replicas, fn.invocation_count
# Deploy a new function
from openfaas_sdk import FunctionDeployment, FunctionResources
spec = FunctionDeployment(
service="env",
image="ghcr.io/openfaas/env:latest",
namespace="openfaas-fn",
labels={"com.openfaas.scale.min": "1"},
limits=FunctionResources(memory="128Mi", cpu="100m"),
)
client.deploy(spec)
# Update an existing function
spec.image = "ghcr.io/openfaas/env:0.2.0"
client.update(spec)
# Delete a function
client.delete_function("env", "openfaas-fn")# List all namespaces
namespaces = client.get_namespaces() # ["openfaas-fn", "staging"]
# Get namespace details
ns = client.get_namespace("openfaas-fn")
# Create a namespace
from openfaas_sdk import FunctionNamespace
client.create_namespace(FunctionNamespace(name="staging", labels={"team": "backend"}))
# Update a namespace
client.update_namespace(FunctionNamespace(name="staging", annotations={"owner": "alice"}))
# Delete a namespace
client.delete_namespace("staging")from openfaas_sdk import Secret
# List secrets
secrets = client.get_secrets("openfaas-fn")
# Create a secret
client.create_secret(Secret(name="db-password", namespace="openfaas-fn", value="s3cr3t"))
# Update a secret
client.update_secret(Secret(name="db-password", namespace="openfaas-fn", value="n3w-s3cr3t"))
# Delete a secret
client.delete_secret("db-password", namespace="openfaas-fn")get_logs returns a lazy iterator that streams NDJSON log lines from the gateway.
# Get the last 100 lines
for msg in client.get_logs("env", "openfaas-fn", tail=100):
print(f"[{msg.timestamp}] {msg.instance}: {msg.text}")
# Follow (stream) logs
for msg in client.get_logs("env", "openfaas-fn", follow=True):
print(msg.text)Filter by time:
from datetime import datetime, timezone
since = datetime(2024, 1, 1, tzinfo=timezone.utc)
for msg in client.get_logs("env", namespace="openfaas-fn", since=since):
print(msg.text)invoke_function returns the raw requests.Response from the function. Unlike gateway API calls, non-2xx status codes are not raised as exceptions, function responses are application-level and it is up to the caller to interpret them.
# POST with a bytes or str payload
resp = client.invoke_function("env", method="POST", payload=b"hello")
if resp.ok:
print(resp.text)
else:
print(f"Error: {resp.status_code} - {resp.text}")
# GET with no payload
resp = client.invoke_function("env", method="GET")
# Custom namespace
resp = client.invoke_function("env", "staging", method="POST")
# Pass extra headers and query parameters
resp = client.invoke_function(
"env",
method="POST",
payload="hello",
headers={"Content-Type": "text/plain"},
query_params={"verbose": "1"},
)invoke_function_async queues the invocation via the gateway's /async-function/ route and returns immediately with a 202 Accepted response. The function result is not returned synchronously.
# Async invocation — returns 202 immediately
client.invoke_function_async("env", payload=b"data")
# Async invocation with a callback URL
client.invoke_function_async(
"env",
payload=b"data",
callback_url="https://my-service.example.com/callback",
)When OpenFaaS IAM is enabled, use use_function_auth=True to automatically obtain a per-function scoped token and attach it as a Bearer token. This requires the client to be configured with a TokenAuth (or an explicit function_token_source):
from openfaas_sdk import Client, TokenAuth, ServiceAccountTokenSource
auth = TokenAuth(
token_url="https://gateway.example.com/oauth/token",
token_source=ServiceAccountTokenSource(),
)
with Client("https://gateway.example.com", auth=auth) as client:
resp = client.invoke_function("my-func", "openfaas-fn", method="POST", use_function_auth=True)
print(resp.text)from openfaas_sdk import Client, BasicAuth
from openfaas.exceptions import NotFoundError, UnauthorizedError, ForbiddenError, APIConnectionError
with Client("https://gateway.example.com", auth=BasicAuth("admin", "secret")) as client:
try:
fn = client.get_function("my-fn", "openfaas-fn")
except NotFoundError:
print("Function does not exist")
except UnauthorizedError:
print("Invalid credentials")
except ForbiddenError:
print("Insufficient permissions")
except APIConnectionError:
print("Could not reach the gateway")| Exception | HTTP status |
|---|---|
NotFoundError |
404 |
UnauthorizedError |
401 |
ForbiddenError |
403 |
UnexpectedStatusError |
any other non-2xx |
APIConnectionError |
network / timeout |
All APIStatusError subclasses expose .status_code and .response (the raw requests.Response).
# Default timeout for all requests (seconds)
client = Client("https://gateway.example.com", auth=auth, timeout=60.0)Pass a pre-configured requests.Session to customise proxies, SSL, or other transport options:
import requests
from openfaas_sdk import Client
session = requests.Session()
session.verify = "/path/to/ca-bundle.pem"
session.proxies = {"https": "http://proxy.corp.example.com"}
client = Client("https://gateway.example.com", auth=auth, http_client=session)When using use_function_auth=True on function invocations, the client needs a token source to obtain per-function scoped tokens. If the client is configured with TokenAuth, it is used automatically. You can also pass a TokenAuth instance explicitly as function_token_source to use a different identity provider for function invocations than the one used for gateway API calls:
client = Client(
"https://gateway.example.com",
function_token_source=TokenAuth(...),
)The FunctionBuilder client interacts with the OpenFaaS Pro Function Builder API to build and push function images from source code.
create_build_context prepares a Docker build context on disk from a template and a handler directory. Templates can be pulled with faas-cli template store pull <lang> or fetched from any other source.
from openfaas.builder import create_build_context, BuildConfig, make_tar, FunctionBuilder
# 1. Assemble the build context from template + handler
context_path = create_build_context(
function_name="hello-world",
handler="./hello-world", # directory containing your function code
language="python3",
template_dir="./template", # directory containing pulled templates
build_dir="./build",
)
# 2. Pack the context into a tar archive
config = BuildConfig(
image="ttl.sh/hello-world:1h",
platforms=["linux/amd64"],
)
make_tar("/tmp/req.tar", context_path, config)build() blocks until the builder returns a complete result:
builder = FunctionBuilder(
"http://127.0.0.1:8081",
hmac_secret="my-hmac-secret", # matches the payload-secret in the cluster
)
result = builder.build("/tmp/req.tar")
print(result.status) # "success" / "failed"
print(result.image) # fully-qualified image name
for line in result.log:
print(line)build_stream() yields BuildResult objects as NDJSON lines arrive, allowing you to display log output in real time:
for result in builder.build_stream("/tmp/req.tar"):
for line in result.log:
print(line)
if result.status in ("success", "failed"):
print("Final status:", result.status)Set skip_push=True on BuildConfig to build the image without pushing it to a registry:
config = BuildConfig(image="ttl.sh/hello-world:1h", skip_push=True)When hmac_secret is provided, every request is signed with an HMAC-SHA256 digest sent in the X-Build-Signature header. The secret must match the payload-secret configured in the builder deployment:
kubectl get secret -n openfaas payload-secret \
-o jsonpath='{.data.payload-secret}' | base64 --decode# Install dependencies
uv sync
# Run tests
uv run python -m pytest -vMIT