Powered byLightning AI
Documentation

    Getting started

Function SDK reference

This reference describes the Python SDK available when you write custom functions in GraphN. Functions run in isolated Firecracker microVMs on Python 3.12 with a configurable memory budget (memory_mb, 256–512 MB, default 512 MB). All SDK methods that perform I/O are async — use await when you call them.

Getting started with functions

Follow these steps to go from an empty function to a working test run:
  1. Open a workspace — Select the workspace where you want the function to live.
  2. Create a function — In the sidebar, click Functions, then create a new function. The editor opens with main.py and requirements.txt.
Function editor showing main.py, requirements.txt, and the test terminal with Run Test button
Function editor showing main.py, requirements.txt, and the test terminal with Run Test button
  1. Define the entry point — In main.py, decorate exactly one async function with @function. Add type hints for each parameter; the workflow passes JSON fields that match those names.
  2. Use the SDK — Call chat, vision, models, kb, storage, and conversion with await (see the sections below).
  3. Save — Use the editor save action so the function version is stored.
  4. Wire the workflow — Open your workflow, add a step that calls this function, and map inputs to the parameter names you defined.
  5. Test — In the test terminal at the bottom of the function editor, enter a JSON object whose keys match your function parameters and click Run Test. The result is the serialized return value of your entry function.

Environment and imports

In the function editor, the platform preloads the @function decorator (alias: tool) and the module instances: chat, vision, models, kb, storage, conversion, veo, nanobanana, and media_tools. You also have access to configure_model and get_model_config for advanced model routing.
You do not need import statements at the top of main.py for these names. If you split code into multiple files in the same function package, import what you need from foundry_helpers (for example from foundry_helpers import kb, storage).
Return JSON-serializable values from your entry function (str, int, float, bool, None, dicts, and lists of those types). The Run tab displays the serialized result.

The @function decorator

Signature

python
def function(func: Callable | None = None, *, timeout: int = 900) -> Callable

Parameters

Scroll horizontally to compare
NameTypeDescription
funcCallable | NoneThe async function being wrapped. Omit when using @function(timeout=…) form.
timeoutintAdvisory timeout in seconds (default 900). Stored as metadata; enforced limits depend on workflow or test harness.

Behavior

  • Exactly one function in your code should be decorated with @function. That function is the entry point.
  • The entry function must be defined with async def.
  • Helper functions can be plain async def without @function.

Examples

python
@function
async def summarize(text: str) -> str:
    return await chat.complete(text, temperature=0.3)
python
@function(timeout=300)
async def quick_task(query: str) -> str:
    return await chat.complete(query, max_tokens=256)

Chat module (chat)

LLM chat completions via an OpenAI-compatible HTTP API. Uses built-in model aliases or custom models registered with configure_model / models.configure.

chat.complete

python
async def complete(
    prompt: str | list[dict[str, Any]],
    model: str | None = None,
    system: str | None = None,
    temperature: float = 0.7,
    max_tokens: int | None = None,
    endpoint: str | None = None,
) -> str
Scroll horizontally to compare
ParameterTypeDescription
promptstr | list[dict]If a string, it becomes a single user message (after optional system). If a list, it must be OpenAI-style messages (role / content).
modelstr | NoneModel alias (for example qwen3-80b) or full model name. See built-in aliases.
systemstr | NoneSystem prompt; only used when prompt is a string.
temperaturefloatSampling temperature (default 0.7).
max_tokensint | NoneMaximum tokens to generate.
endpointstr | NoneOverride the inference base URL (optional).
Returns: str — assistant message content.
Example:
python
reply = await chat.complete("Explain async/await in one sentence.", temperature=0.2)

Vision module (vision)

Image understanding via a vision-language model. Images may be URLs, data URLs, or raw base64 (non-URL strings are treated as JPEG base64).

vision.analyze

python
async def analyze(
    image: str | list[str],
    prompt: str,
    model: str | None = None,
    temperature: float = 0.3,
    max_tokens: int | None = None,
    endpoint: str | None = None,
) -> str
Scroll horizontally to compare
ParameterTypeDescription
imagestr | list[str]One image or several.
promptstrQuestion or instruction for the model.
modelstr | NoneModel alias (for example qwen3-vl) or full name.
temperaturefloatSampling temperature (default 0.3).
max_tokensint | NoneCap on generated tokens.
endpointstr | NoneOverride the inference base URL.
Returns: str — model text about the image(s).
Example:
python
text = await vision.analyze("https://example.com/photo.jpg", "List the main objects.")

vision.describe

python
async def describe(
    image: str,
    model: str | None = None,
    endpoint: str | None = None,
) -> str
Scroll horizontally to compare
ParameterTypeDescription
imagestrImage URL, data URL, or base64.
modelstr | NoneVision model alias or name.
endpointstr | NoneOverride inference base URL.
Returns: str — detailed description.
Example:
python
caption = await vision.describe(image_url)

vision.extract_text

python
async def extract_text(
    image: str,
    model: str | None = None,
    endpoint: str | None = None,
) -> str
Returns: str — OCR-style extracted text (uses a low temperature internally).
Example:
python
ocr = await vision.extract_text(scan_image_b64)

Models module (models) and global helpers

Built-in aliases

The function runtime resolves model aliases through the platform model registry (app/models/config.yaml). Commonly used built-in aliases:
Scroll horizontally to compare
AliasTypeNotes
qwen3-80bChatDefault chat model; same as the agent model picker's Qwen3 80B Instruct (131K context).
qwen3-235bChatLargest MoE chat model (262K context) for the hardest reasoning and long-context tasks.
qwen3-coderChatCode-specialized MoE model for agentic software-engineering workflows (262K context).
qwen3-vlVisionDefault vision model; same as the agent model picker's Qwen3.5 9B Vision (image+video).
qwen3.8-27bChat/VisionDense 27B VL for harder coding and agentic work; Qwen3.8 27B (image+video).
This is not the complete set — your deployment also registers models such as gpt-oss-120b, gemma-4, and nemotron-3-super, plus any models you have imported. Call models.list_available() for the live list, or see the Models reference for the full picker. You can register additional aliases with configure_model or models.configure.

configure_model (module-level)

python
def configure_model(alias: str, endpoint: str, model_name: str) -> None
Scroll horizontally to compare
ParameterTypeDescription
aliasstrShort name used in model= arguments.
endpointstrOpenAI-compatible API base (for example https://host/v1).
model_namestrModel identifier expected by that endpoint.
Returns: None
Example:
python
configure_model("my-llm", "https://inference.example/v1", "vendor/model-id")
out = await chat.complete("Hi", model="my-llm")

get_model_config

python
def get_model_config(model: str | None = None, default_type: str = "chat") -> tuple[str, str]
Scroll horizontally to compare
ParameterTypeDescription
modelstr | NoneAlias or name; None uses defaults for default_type.
default_typestr"chat" or "vision" when resolving defaults.
Returns: tuple[str, str] — (endpoint, model_name).
Example:
python
endpoint, name = get_model_config("qwen3-80b", "chat")

models.list_available

python
def list_available(type_filter: str | None = None) -> list[str]
Scroll horizontally to compare
ParameterTypeDescription
type_filterstr | NonePass "chat" or "vision" to narrow the list.
Returns: list[str] — known aliases (built-ins plus registered names).
Example:
python
names = models.list_available("chat")

models.get_default

python
def get_default(model_type: str = "chat") -> str | None
Returns: str | None — default alias for "chat" or "vision", or None if unavailable.
Example:
python
default_chat = models.get_default("chat")

models.configure

python
def configure(alias: str, endpoint: str, model_name: str) -> None
Same behavior as configure_model.
Example:
python
models.configure("fast", "http://localhost:8000/v1", "my-model")

models.get_config

python
def get_config(model: str | None = None) -> dict[str, str]
Returns: dict with keys endpoint and model.
Example:
python
cfg = models.get_config("qwen3-vl")

Knowledge base module (kb)

Search, embeddings, and document lifecycle for workspace knowledge bases. IDs look like kb_….

kb.search

python
async def search(
    kb_id: str,
    query: str,
    top_k: int = 5,
    rerank: bool = True,
    reranker_model: str | None = None,
) -> list[dict[str, Any]]
Scroll horizontally to compare
ParameterTypeDescription
kb_idstrKnowledge base ID.
querystrSearch query.
top_kintNumber of hits (default 5).
rerankboolWhether to rerank for relevance.
reranker_modelstr | NoneReranker name; use "none" to skip reranking.
Returns: list[dict] — results (typically include text, score, and metadata fields).
Example:
python
hits = await kb.search("kb_abc123", "refund policy", top_k=3)

kb.embed

python
async def embed(text: str, normalize: bool = True) -> list[float]
Returns: list[float] — one 1024-dimensional embedding.
Example:
python
vec = await kb.embed("hello world")

kb.embed_batch

python
async def embed_batch(texts: list[str], normalize: bool = True) -> list[list[float]]
Scroll horizontally to compare
ParameterTypeDescription
textslist[str]Up to 100 strings per request.
Returns: list[list[float]] — one vector per input string.
Example:
python
vectors = await kb.embed_batch(["a", "b"])

kb.list_kbs

python
async def list_kbs(
    *,
    limit: int | None = None,
    cursor: str | None = None,
) -> list[dict] | dict
When called without cursor, returns a plain list[dict] of all knowledge bases (backward compatible). When cursor is provided (pass "" for the first page), returns a paginated envelope: {"items": [...], "has_more": bool, "next_cursor": "..."}. When has_more is false, next_cursor is omitted. Pass next_cursor as cursor to fetch the next page.
Example:
python
all_kbs = await kb.list_kbs()

# Paginated
page = await kb.list_kbs(limit=10, cursor="")
for kb_item in page["items"]:
    print(kb_item["name"])

kb.list_kbs_all

python
async def list_kbs_all(*, page_size: int = 50) -> list[dict[str, Any]]
Returns: list[dict] — all knowledge bases, auto-paginating through all pages.
Example:
python
all_kbs = await kb.list_kbs_all()

kb.create

python
async def create(name: str, description: str = "") -> dict[str, Any]
Returns: dict — created KB fields (includes id, name, description, created_at).
Example:
python
info = await kb.create("Support FAQs", "Indexed answers")

kb.get

python
async def get(kb_id: str) -> dict[str, Any]
Returns: dict — includes id, name, description, document_count, chunk_count.
Example:
python
meta = await kb.get("kb_abc123")

kb.upload_document

python
async def upload_document(
    kb_id: str,
    content: str | bytes,
    filename: str,
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]
Scroll horizontally to compare
ParameterTypeDescription
contentstr | bytesRaw document body; chunked and embedded by the service.
filenamestrFilename (used for type detection, for example .md, .txt).
metadatadict | NoneOptional JSON-serializable metadata.
Returns: dict — includes document_id, chunk_count, filename (and related fields).
Example:
python
await kb.upload_document("kb_abc123", "# Title\n\nBody", "notes.md")

kb.list_documents

python
async def list_documents(
    kb_id: str,
    *,
    limit: int | None = None,
    cursor: str | None = None,
) -> list[dict] | dict
When called without cursor, returns a plain list[dict] (backward compatible). With cursor, returns {"items": [...], "has_more": bool, "next_cursor": "..."}.
Returns: list[dict] or paginated envelope — documents with id, filename, metadata, chunk_count, etc.
Example:
python
docs = await kb.list_documents("kb_abc123")

kb.batch_get

python
async def batch_get(ids: list[str]) -> dict
Returns: dict — {"total": N, "succeeded": N, "failed": N, "results": [...]}.
Each item in results has {"id": "...", "success": bool}. On success, a "knowledgebase" key contains the full KB dict. On failure, an "error" key contains {"code": "not_found"|"forbidden"|..., "message": "..."}.

kb.batch_update

python
async def batch_update(items: list[dict]) -> dict
Each item must contain "id" and any updatable fields ("name", "description", "chunk_size", "chunk_overlap").
Returns: dict — {"total": N, "succeeded": N, "failed": N, "results": [...]}.
Example:
python
result = await kb.batch_update([
    {"id": "kb_1", "name": "Renamed KB"},
    {"id": "kb_2", "description": "Updated desc"},
])

kb.batch_delete

python
async def batch_delete(ids: list[str]) -> dict
Permanently deletes each KB via the same cascade as kb.delete. Vectors, document metadata, ingest bookkeeping, and the KB record are required cleanup; cancellation of in-flight ingest and media-object cleanup are best-effort. There is no soft-delete. Partial success is possible — check results.
Returns: dict — {"total": N, "succeeded": N, "failed": N, "results": [...]}.
Example:
python
result = await kb.batch_delete(["kb_1", "kb_2"])

kb.batch_get_documents

python
async def batch_get_documents(kb_id: str, ids: list[str]) -> dict
Returns: dict — {"total": N, "succeeded": N, "failed": N, "results": [...]}.
Each item in results has {"id": "...", "success": bool}. On success, a "document" key contains the full document dict. On failure, an "error" key contains {"code": "...", "message": "..."}.

kb.batch_delete_documents

python
async def batch_delete_documents(kb_id: str, ids: list[str]) -> dict
Returns: dict — {"total": N, "succeeded": N, "failed": N, "results": [...]}.

kb.ingest_from_storage

python
async def ingest_from_storage(
    kb_id: str,
    storage_id: str,
    file_path: str,
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]
Downloads an object from a storage bucket and uploads it to the KB. Skips duplicate ingest when the same source_path (or matching original_path in metadata) already exists.
Returns: dict — includes document_id, chunk_count, filename, source_path, and skipped: True when deduplicated.
Example:
python
await kb.ingest_from_storage("kb_abc123", "my-store", "exports/doc.md")

kb.delete

python
async def delete(kb_id: str) -> bool
Permanently deletes the knowledge base. There is no soft-delete or restore. A successful call purges all vectors and document metadata, removes ingest bookkeeping, and deletes the KB record. Cancellation of in-flight async ingest and cleanup of media objects under the KB prefix are best-effort.
Also available from the UI (Knowledge Bases), CLI (graphn kb delete <id> [--force]), and MCP (graphn_kb_delete).
Returns: bool — True on success (HTTP 200 or 204).
Example:
python
await kb.delete("kb_abc123")

Storage module (storage)

Object storage scoped to your workspace. Use storage IDs (bucket names) and keys (object paths). Calls use the platform-injected credentials; you do not configure API keys in code.

storage.upload

python
async def upload(
    storage_id: str,
    key: str,
    content: str | bytes,
    content_type: str | None = None,
) -> dict[str, Any]
Returns: dict — key, size, storage_id, etag.
Example:
python
await storage.upload("my-store", "out/report.md", "# Report\n", content_type="text/markdown")

storage.upload_file

python
async def upload_file(
    storage_id: str,
    key: str,
    local_path: str,
    content_type: str | None = None,
) -> dict[str, Any]
Returns: Same shape as upload.
Example:
python
await storage.upload_file("my-store", "data/file.bin", "/tmp/file.bin")

storage.download

python
async def download(storage_id: str, key: str) -> bytes
Returns: bytes — object body.
Example:
python
data = await storage.download("my-store", "out/report.md")
text = data.decode("utf-8")

storage.download_file

python
async def download_file(storage_id: str, key: str, local_path: str) -> None
Writes bytes to local_path. Returns: None
Example:
python
await storage.download_file("my-store", "out/report.md", "/tmp/report.md")

storage.list

python
async def list(
    storage_id: str,
    prefix: str = "",
    max_keys: int = 1000,
) -> list[dict[str, Any]]
Returns: list[dict] — each item includes key, size, last_modified, etag.
Example:
python
files = await storage.list("my-store", prefix="out/")

storage.list_objects

python
async def list_objects(
    storage_id: str,
    prefix: str = "",
    max_keys: int = 10000,
) -> list[dict[str, Any]]
Paginates until all matching keys are retrieved (up to max_keys total).
Returns: list[dict] — file records from the API.
Example:
python
all_files = await storage.list_objects("my-store", prefix="chat/")

storage.delete

python
async def delete(storage_id: str, key: str) -> bool
Deletes the object; missing objects do not cause failure. Returns: True
Example:
python
await storage.delete("my-store", "tmp/scratch.dat")

storage.presigned_url

python
async def presigned_url(
    storage_id: str,
    key: str,
    expires_in: int = 3600,
    upload: bool = False,
) -> str
Scroll horizontally to compare
ParameterTypeDescription
expires_inintLifetime in seconds (default one hour).
uploadboolTrue for a presigned upload (PUT) URL.
Returns: str — URL string.
Example:
python
url = await storage.presigned_url("my-store", "public/file.pdf")

storage.copy

python
async def copy(storage_id: str, source_key: str, dest_key: str) -> dict[str, Any]
Returns: dict — includes key and storage_id for the destination.
Example:
python
await storage.copy("my-store", "a.txt", "archive/a.txt")

storage.exists

python
async def exists(storage_id: str, key: str) -> bool
Returns: bool — whether an object exists at key.
Example:
python
if await storage.exists("my-store", "out/ready.flag"):
    ...

storage.create

python
async def create(name: str, description: str = "") -> dict[str, Any]
Creates a new bucket. Returns: dict — name, status, etc.
Example:
python
await storage.create("project-files", "Pipeline artifacts")

storage.list_stores

python
async def list_stores() -> list[str]
Returns: list[str] — store names. Each name is the storage_id used by upload/download/create. Do not treat items as dicts. For status, description, and object counts, use list_store_details().
Example:
python
stores = await storage.list_stores()  # ["default", "production-studio"]
if "default" not in stores:
    await storage.create("default", description="")

storage.list_store_details

python
async def list_store_details() -> list[dict[str, Any]]
Returns: list[dict] — each record has id (always equal to name, the storage_id), plus display_name, description, status, object_count, and total_size_bytes.
Example:
python
details = await storage.list_store_details()
# [{id, name, display_name, description, status, object_count, total_size_bytes}, ...]

storage.list_storages

python
async def list_storages() -> list[dict[str, Any]]
Deprecated alias of list_store_details(). Returns: the same records (id equals name, plus status/description/counts).
Example:
python
rows = await storage.list_storages()

Conversion module (conversion)

PDF and image conversion to markdown via the document conversion service. Source files must live in storage; results are written back to storage and optionally ingested into a KB.

conversion.convert_document

python
async def convert_document(
    storage_id: str,
    file_path: str,
    ocr_engine: str = "auto",
    start_page: int = 1,
    end_page: int | None = None,
) -> dict[str, Any]
Scroll horizontally to compare
ParameterTypeDescription
storage_idstrBucket containing the file.
file_pathstrObject key (PDF, TIFF, or image).
ocr_enginestr"auto" (default), "deepseek", "hunyuan", or "hybrid". "digital" and "scanned" are still accepted as legacy aliases.
start_pageintFirst page (1-based).
end_pageint | NoneLast page, or None for all.
Returns: dict with markdown, markdown_path, pages_processed, ocr_engine, usage, plus storage_id and source_path.
Example:
python
result = await conversion.convert_document("my-store", "uploads/spec.pdf", ocr_engine="auto")
md = result["markdown"]

conversion.convert_and_ingest

python
async def convert_and_ingest(
    storage_id: str,
    file_path: str,
    kb_id: str,
    ocr_engine: str = "auto",
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]
Converts the file, then uploads the markdown to kb_id.
Returns: dict — includes markdown_path, kb_id, document_id, chunk_count, pages_processed, ocr_engine, storage_id, source_path.
Example:
python
await conversion.convert_and_ingest("my-store", "scan.pdf", "kb_abc123")

Error handling

  • SDK methods raise standard Python exceptions when something goes wrong — for example httpx.HTTPStatusError from failed HTTP calls, ValueError for invalid IDs or empty uploads, FileNotFoundError when a storage object is missing, RuntimeError or ConnectionError for service errors, and TimeoutError when polling jobs exceed their limits.
  • Catch exceptions you can handle and return a clear message or fallback result.
  • Unhandled exceptions propagate to the platform: the function step fails, and the error message is surfaced in the run output.

Pip dependencies

List third-party packages in requirements.txt (one package per line, standard pip syntax). The platform installs these when building the function environment. Only add libraries you need; prefer the built-in SDK for LLM, vision, KB, storage, and conversion flows when possible.

Additional modules

The following sections cover veo, nanobanana, and media_tools. Prefer these over any older video / image helper names.

Veo 3.1 video generation (veo)

Generate videos with native audio using Google Veo 3.1. The model produces synchronized speech and sound directly -- no separate TTS step needed. The module proxies through the backend so your function never touches GCP credentials.

veo.generate(prompt, *, aspect_ratio, duration, resolution, person_generation, storage_id)

Generate a video from a text prompt.
Scroll horizontally to compare
ParameterTypeDefaultDescription
promptstr(required)Text description of the video
aspect_ratiostr"16:9""16:9" (landscape) or "9:16" (portrait)
durationint8Clip length in seconds (4 to 8)
resolutionstr"720p""720p" or "1080p"
person_generationstr"allow_adult""allow_adult", "dont_allow", or "allow_all"
storage_idstr""Workspace storage to upload result to
Returns a dict with video_id, status, storage_path, and storage_id. Pass storage_path (with storage_id) to veo.extend to chain another segment.
python
result = await veo.generate("A woman presenting a product on camera", resolution="1080p")
print(result["storage_path"])  # generated/videos/veo/<exec_id>/<hash>.mp4

veo.generate_from_image(prompt, image_source, *, ...)

Generate a video using a reference image as the first frame (image-to-video). Same parameters as generate plus image_source (local path or storage path).
python
avatar = await nanobanana.generate("professional woman, business attire")
result = await veo.generate_from_image(
    "The woman speaks to the camera about quarterly results",
    avatar["storage_path"],
    storage_id=avatar["storage_id"],
)

veo.extend(prompt, video_source, *, resolution, person_generation, storage_id)

Extend an existing video by approximately 7 seconds. Each call produces a continuation from the last second of the input video. Use this to checkpoint long videos: if any single extension fails or your function is interrupted, you only lose the in-flight ~7s, not the entire chain.
Scroll horizontally to compare
ParameterTypeDefaultDescription
promptstr(required)Text prompt guiding the extension
video_sourcestr(required)Storage path returned by a previous generate or extend call (storage_path field), or a local file path to an mp4
resolutionstr"720p""720p" or "1080p"
person_generationstr"allow_adult"Person generation policy
storage_idstr""Workspace storage that contains video_source (also where the new segment is uploaded). Required when video_source is a storage path
Returns a dict with video_id, status, storage_path, and storage_id. Pass storage_path back to veo.extend to keep chaining.
python
initial = await veo.generate("A woman speaking to camera")
extended = await veo.extend(
    "She continues explaining the topic",
    initial["storage_path"],
    storage_id=initial["storage_id"],
)

veo.generate_long(prompt, *, target_duration, image_base64, aspect_ratio, resolution, person_generation, storage_id)

Generate a long video by automatically chaining an initial generation with multiple extensions. Maximum duration is approximately 148 seconds (8s initial + 20 extensions of 7s each).
Scroll horizontally to compare
ParameterTypeDefaultDescription
promptstr(required)Text prompt for the entire video
target_durationint30Desired length in seconds (5-148)
image_base64str | NoneNoneOptional base64-encoded first-frame image
aspect_ratiostr"16:9""16:9" or "9:16"
resolutionstr"720p""720p" or "1080p"
person_generationstr"allow_adult"Person generation policy
storage_idstr""Workspace storage to upload result to
Returns a dict with video_id, status, storage_path, storage_id, segment_paths (one storage path per segment, in order -- pass any to veo.extend to fork from that segment), segments, and estimated_duration.
python
long_video = await veo.generate_long(
    "Professional woman delivering a quarterly earnings presentation",
    target_duration=30,
    resolution="1080p",
)
print(f"Generated {long_video['segments']} segments, ~{long_video['estimated_duration']}s")

Avatar image generation (nanobanana)

Generate avatar and character images using Nano Banana (Gemini Flash Image). Useful for creating reference images to feed into veo.generate_from_image.

nanobanana.generate(description, style, storage_id)

Scroll horizontally to compare
ParameterTypeDefaultDescription
descriptionstr(required)Text description of the avatar
stylestr"realistic"Art style: "realistic", "anime", "3d", "pixel-art", etc.
storage_idstr""Workspace storage to upload result to
Returns a dict with image_id, status, storage_path, storage_id.
python
avatar = await nanobanana.generate(
    "professional woman in business attire, neutral background, front-facing portrait",
    style="realistic",
)

Media tools (media_tools)

Low-level media manipulation via the graphn-media FFmpeg service. Useful for replacing or mixing audio tracks on existing video files.

media_tools.replace_audio(video_source, audio_source, storage_id)

Replace a video's audio track with a different audio file. The output length matches the shorter of the two inputs.
Scroll horizontally to compare
ParameterTypeDefaultDescription
video_sourcestr(required)Local path or storage path to the video file
audio_sourcestr(required)Local path or storage path to the audio file
storage_idstr""Workspace storage to upload result to
python
final = await media_tools.replace_audio(
    "videos/original.mp4",
    "audio/soundtrack.wav",
    storage_id="default",
)
Previous

API reference

Next

Models