SnakeBridge

Compile-time generator for type-safe Elixir bindings to Python libraries.

Installation

Add SnakeBridge to your dependencies and configure Python libraries in your mix.exs:

defmodule MyApp.MixProject do
  use Mix.Project

  def project do
    [
      app: :my_app,
      version: "1.0.0",
      elixir: "~> 1.14",
      deps: deps(),
      python_deps: python_deps(),
      # IMPORTANT: Add the snakebridge compiler for Python introspection
      compilers: [:snakebridge] ++ Mix.compilers()
    ]
  end

  defp deps do
    [
      {:snakebridge, "~> 0.9.0"}
    ]
  end

  # Define Python dependencies just like Elixir deps
  defp python_deps do
    [
      {:numpy, "1.26.0"},
      {:pandas, "2.0.0", include: ["DataFrame", "read_csv"]}
    ]
  end
end

Important: The compilers: [:snakebridge] ++ Mix.compilers() line is required for:

• Automatic Python package installation at compile time
• Type introspection and wrapper generation
• Keeping packages up-to-date when requirements change

The python_deps function mirrors how deps works - a list of tuples with the library name, version, and optional configuration.

Then add runtime configuration in config/runtime.exs:

import Config

# Auto-configure snakepit for snakebridge
SnakeBridge.ConfigHelper.configure_snakepit!()

Quick Start

# Generated wrappers work like native Elixir
{:ok, result} = Numpy.mean([1, 2, 3, 4])

# Optional Python arguments via keyword opts
{:ok, result} = Numpy.mean([[1, 2], [3, 4]], axis: 0)

# Runtime flags (idempotent caching, timeouts)
{:ok, result} = Numpy.mean([1, 2, 3], idempotent: true)

Features

Generated Wrappers

SnakeBridge generates Elixir modules that wrap Python libraries:

# Python: numpy.mean(a, axis=None, dtype=None, keepdims=False)
# Generated: Numpy.mean(a, opts \\ [])

Numpy.mean([1, 2, 3])                      # Basic call
Numpy.mean([1, 2, 3], axis: 0)             # With Python kwargs
Numpy.mean([1, 2, 3], idempotent: true)    # With runtime flags
Numpy.reshape([1, 2, 3, 4], [[2, 2]], order: "C") # Optional kwargs in args slot

All wrappers accept:

• Extra positional args: args list appended after required parameters
• Keyword options: opts for Python kwargs and runtime flags (idempotent, __runtime__)

If a function has optional positional parameters, the generated wrapper includes an args list. Pass [] when you do not need extra positional args, or pass a keyword list directly as the args argument.

Signature & Arity Model

SnakeBridge matches call-site arity against a manifest range so optional args and keyword opts do not produce perpetual "missing" symbols. Required keyword-only parameters are documented and validated at runtime.

When a Python signature is unavailable (common for C-extensions), SnakeBridge generates variadic wrappers with convenience arities up to a configurable max (default 8):

config :snakebridge, variadic_max_arity: 8

Python function and method names that are invalid in Elixir are sanitized (for example, class → py_class). The manifest stores the Python↔Elixir mapping and runtime calls use the original Python name. Common dunder methods map to idiomatic names (for example, __init__ → new, __len__ → length).

Class Constructors

Classes generate new/N matching their Python __init__:

# Python: class Point:
#           def __init__(self, x, y): ...
# Generated: Geometry.Point.new(x, y, opts \\ [])

{:ok, point} = Geometry.Point.new(10, 20)
{:ok, x} = Geometry.Point.x(point)  # Attribute access
:ok = SnakeBridge.Runtime.release_ref(point)

Class vs Submodule Resolution

Nested calls like Lib.Foo.bar/… are resolved automatically via introspection. SnakeBridge checks whether Foo is a class attribute on the parent module first, and falls back to a submodule when it is not. This means classes are detected without manual include.

Instance Attributes

Read and write Python object attributes:

# Get attribute
{:ok, value} = SnakeBridge.Runtime.get_attr(instance, "attribute_name")

# Set attribute
:ok = SnakeBridge.Runtime.set_attr(instance, "attribute_name", new_value)

Module Attributes

Module-level constants and objects are exposed via generated zero-arity accessors or the runtime API:

{:ok, pi} = Math.pi()
{:ok, nan} = Numpy.nan()

# Or via the runtime helper
{:ok, pi} = SnakeBridge.Runtime.get_module_attr(Math, :pi)

Dynamic Dispatch (No-Codegen)

Call Python functions and methods without generated wrappers:

# Call a function by module path
{:ok, value} = SnakeBridge.Runtime.call_dynamic("math", "sqrt", [144])

# Create a ref and call methods dynamically
{:ok, ref} = SnakeBridge.Runtime.call_dynamic("pathlib", "Path", ["."])
{:ok, exists?} = SnakeBridge.Dynamic.call(ref, :exists, [])
{:ok, name} = SnakeBridge.Dynamic.get_attr(ref, :name)
{:ok, _} = SnakeBridge.Dynamic.set_attr(ref, :name, "snakebridge")

Use generated wrappers when you want compile-time arity checks, docs, and faster hot-path calls. Use dynamic dispatch when symbols are discovered at runtime or when introspection cannot see the module/class ahead of time.

Performance considerations: dynamic calls are string-based and skip codegen optimizations, so prefer generated wrappers for frequently called functions.

Session Lifecycle Management

SnakeBridge scopes Python object refs to sessions and releases them when the owning process exits. Use SessionContext.with_session/1 to bind a session to the current process:

SnakeBridge.SessionContext.with_session(fn ->
  {:ok, ref} = SnakeBridge.Runtime.call_dynamic("pathlib", "Path", ["."])
  {:ok, exists?} = SnakeBridge.Dynamic.call(ref, :exists, [])
end)

Refs created inside the block share the same session_id, so multiple calls can chain on the same Python objects. The SessionManager monitors the owner process and calls Python release_session when it dies. You can also release explicitly:

:ok = SnakeBridge.SessionManager.release_session(session_id)

Session cleanup logs are opt-in. To enable them, set:

config :snakebridge, session_cleanup_log_level: :debug

SnakeBridge also emits [:snakebridge, :session, :cleanup] telemetry events on cleanup.

Pass explicit options with SessionContext.with_session/2 when you need a custom session id or metadata:

SnakeBridge.SessionContext.with_session(session_id: "analytics", fn ->
  {:ok, ref} = SnakeBridge.Runtime.call_dynamic("pathlib", "Path", ["."])
  {:ok, _} = SnakeBridge.Dynamic.call(ref, :exists, [])
end)

Configuration options (environment variables):

• SNAKEBRIDGE_REF_TTL_SECONDS (default 0, disabled) to enable time-based cleanup
• SNAKEBRIDGE_REF_MAX to cap in-memory refs per Python process

Session Affinity (Snakepit)

SnakeBridge relies on Snakepit's session affinity to route calls with the same session_id to the same worker. By default, affinity is a hint: under load, Snakepit may route a session to a different worker, which can invalidate in-memory refs.

For strict routing of stateful refs, configure affinity at the pool level or per call:

# Pool default (recommended for stateful refs)
SnakeBridge.ConfigHelper.configure_snakepit!(affinity: :strict_queue)

# Per-call override
SnakeBridge.call("pathlib", "Path", ["."],
  __runtime__: [affinity: :strict_fail_fast]
)

Strict modes:

• :strict_queue — queue until the preferred worker is available
• :strict_fail_fast — return {:error, :worker_busy} when the preferred worker is busy

See guides/SESSION_AFFINITY.md for routing details, error semantics, and streaming guidance.

Multi-Session: Multiple Snakes in the Pit

SnakeBridge supports running multiple isolated Python sessions concurrently. Each session maintains independent state - objects created in one session are invisible to others. This enables:

• Multi-tenant isolation: Each user/tenant gets their own Python state
• Parallel processing: Workers with isolated state, no cross-contamination
• A/B testing: Different configurations running side-by-side
• Resource isolation: Separate memory pools per session

Concurrent Sessions

# Two sessions running concurrently with independent state
tasks = [
  Task.async(fn ->
    SessionContext.with_session(session_id: "tenant_a", fn ->
      {:ok, ref} = SnakeBridge.call("collections", "Counter", [["a", "b", "a"]])
      # This Counter exists only in tenant_a's session
      SnakeBridge.Dynamic.call(ref, :most_common, [1])
    end)
  end),
  Task.async(fn ->
    SessionContext.with_session(session_id: "tenant_b", fn ->
      {:ok, ref} = SnakeBridge.call("collections", "Counter", [["x", "y", "x"]])
      # Completely isolated from tenant_a
      SnakeBridge.Dynamic.call(ref, :most_common, [1])
    end)
  end)
]

[result_a, result_b] = Task.await_many(tasks)

Session Isolation Guarantee

Objects (refs) are scoped to their session. A ref created in session A cannot be accessed from session B:

ref_a = SessionContext.with_session(session_id: "session_a", fn ->
  {:ok, ref} = SnakeBridge.call("pathlib", "Path", ["/data/a"])
  ref  # ref.session_id == "session_a"
end)

# ref_a belongs to session_a - it carries its session_id
IO.puts(ref_a.session_id)  # => "session_a"

Parallel Processing Pattern

Process items in parallel with isolated sessions per worker:

items
|> Task.async_stream(fn item ->
  SessionContext.with_session(session_id: "worker_#{item.id}", fn ->
    # Each worker has fully isolated Python state
    {:ok, result} = SnakeBridge.call("json", "dumps", [item.data])
    result
  end)
end, max_concurrency: System.schedulers_online())
|> Enum.map(fn {:ok, result} -> result end)

See examples/multi_session_example for complete working code demonstrating multi-session patterns and affinity modes under load.

Streaming Functions

Configure streaming functions to generate *_stream variants:

# In mix.exs
{:llm, version: "1.0", streaming: ["generate", "complete"]}

# Generated variants:
LLM.generate(prompt)                              # Returns complete result
LLM.generate_stream(prompt, opts, callback)       # Streams chunks to callback

# Usage:
LLM.generate_stream("Hello", [], fn chunk ->
  IO.write(chunk)
end)

Streaming calls use Snakepit's server-side streaming RPC BridgeService.ExecuteStreamingTool. Tools must be registered with supports_streaming: true for streaming to work; the ExecuteToolRequest.stream field alone is not sufficient.

Generators and Iterators

Python generators and iterators are returned as SnakeBridge.StreamRef and implement the Enumerable protocol for lazy iteration:

{:ok, stream} = SnakeBridge.Runtime.call_dynamic("itertools", "count", [1])
Enum.take(stream, 5)

Performance considerations: each element is fetched over the runtime boundary. Prefer batching (e.g., Python-side list construction) for large iterations, and use bounded Enum operations (Enum.take/2, Enum.reduce/3) to limit round-trips.

Protocol Integration (Refs)

Python refs implement Elixir protocols for smoother interop:

{:ok, ref} = SnakeBridge.Runtime.call_dynamic("builtins", "range", [0, 3])

inspect(ref)          # Uses Python __repr__ / __str__
"Range: #{ref}"       # Uses Python __str__
Enum.count(ref)       # Calls __len__
Enum.map(ref, &(&1 * 2))

Python Context Managers

Use SnakeBridge.with_python/2 to safely call __enter__ and __exit__:

{:ok, file} = SnakeBridge.Runtime.call_dynamic("builtins", "open", ["output.txt", "w"])

SnakeBridge.with_python(file) do
  SnakeBridge.Dynamic.call(file, :write, ["hello\\n"])
end

The context variable inside the block is bound to the __enter__ return value.

Callbacks (Elixir → Python)

Elixir functions can be passed to Python as callbacks:

callback = fn x -> x * 2 end
{:ok, stream} = SnakeBridge.Runtime.call_dynamic("builtins", "map", [callback, [1, 2, 3]])
Enum.to_list(stream)

Performance considerations: callbacks cross the boundary per invocation. Keep callback work small or batch on the Python side when possible.

Strict Mode for CI

Enable strict mode to verify generated code integrity:

# In CI
SNAKEBRIDGE_STRICT=1 mix compile

Strict mode verifies:

1. All used symbols are in the manifest
2. All generated files exist
3. Expected functions are present in generated files

Documentation Conversion

Python docstrings are converted to ExDoc Markdown:

• NumPy style -> Markdown sections
• Google style -> Markdown sections
• Sphinx/Epytext styles supported
• RST math (:math:`E=mc^2) -> KaTeX ($E=mc^2$)

Wire Schema (v1)

SnakeBridge tags non-JSON values with __type__ and __schema__ markers to keep the Elixir/Python contract stable across versions. Atoms are encoded as tagged values and decoded only when allowlisted:

config :snakebridge, atom_allowlist: ["ok", "error"]

Python decodes tagged atoms to plain strings by default for compatibility with most libraries. Opt in to Atom wrapper objects by setting:

SNAKEBRIDGE_ATOM_CLASS=true

Python results that are not JSON-serializable are automatically returned as refs (e.g., {"__type__": "ref", ...}) so you can chain method calls on the returned object. Each ref includes a session_id to keep ownership scoped to the calling process.

Graceful Serialization

SnakeBridge preserves container structure when encoding Python results. If a list or dict contains non-serializable objects, only those specific items become refs - the rest of the container remains accessible as normal Elixir data.

This is especially useful for DSPy-like "history" structures where most fields are serializable but some (like response objects) are not:

{:ok, history} = SnakeBridge.call("dspy_module", "get_history", [])

# history is a list of maps - NOT a single opaque ref
for entry <- history do
  # Serializable fields are directly accessible
  IO.puts("Model: #{entry["model"]}, cost: #{entry["cost"]}")

  # Non-serializable fields become nested refs
  response = entry["response"]
  if SnakeBridge.ref?(response) do
    # Can still access the ref's attributes
    {:ok, id} = SnakeBridge.attr(response, "id")
    IO.puts("Response ID: #{id}")
  end
end

Key behaviors:

• Container preservation: Lists and dicts keep their structure; only non-serializable leaves become %SnakeBridge.Ref{} or %SnakeBridge.StreamRef{}
• Cycle detection: Self-referential structures are handled safely - cycles become refs
• Type metadata: Refs include type_name for inspection (the Python class name)

For Snakepit-style unserializable markers (used by Snakepit's internal adapters, not SnakeBridge encoding), see SnakeBridge.unserializable?/1 and SnakeBridge.unserializable_info/1.

ML Error Translation

Python ML exceptions are translated to structured Elixir errors:

# Shape mismatches with tensor dimensions
%SnakeBridge.Error.ShapeMismatchError{expected: [3, 4], actual: [4, 3]}

# Out of memory with device info
%SnakeBridge.Error.OutOfMemoryError{device: :cuda, available: 1024, requested: 2048}

# Dtype conflicts with casting guidance
%SnakeBridge.Error.DtypeMismatchError{expected: :float32, actual: :float64}

Use SnakeBridge.ErrorTranslator.translate/1 for manual translation, or set error_mode to translate on every runtime call:

config :snakebridge, error_mode: :translated

Unknown Python exceptions are mapped dynamically into SnakeBridge.DynamicException.* modules so you can rescue by type:

config :snakebridge, error_mode: :raise_translated

try do
  SnakeBridge.Runtime.call_dynamic("builtins", "int", ["not-a-number"])
rescue
  e in SnakeBridge.DynamicException.ValueError ->
    IO.puts("Caught: #{Exception.message(e)}")
end

Telemetry

The compile pipeline emits telemetry events:

# Attach handler
:telemetry.attach("my-handler", [:snakebridge, :compile, :stop], fn _, measurements, _, _ ->
  IO.puts("Compiled #{measurements.symbols_generated} symbols")
end, nil)

Compile events:

• [:snakebridge, :compile, :start|:stop|:exception]
• [:snakebridge, :compile, :scan, :stop]
• [:snakebridge, :compile, :introspect, :start|:stop]
• [:snakebridge, :compile, :generate, :stop]
• [:snakebridge, :docs, :fetch]
• [:snakebridge, :lock, :verify]

Runtime events (forwarded from Snakepit):

• [:snakebridge, :runtime, :call, :start|:stop|:exception]

Script shutdown events (forwarded from Snakepit when attached):

• [:snakebridge, :script, :shutdown, :start|:stop|:cleanup|:exit] Enable via SnakeBridge.Telemetry.ScriptShutdownForwarder.attach().

Telemetry metadata schema:

• Compile events include library, phase, and details.
• Runtime events include library, function, and call_type.

Breaking change: compile phase events now live under [:snakebridge, :compile, ...] and share the unified metadata schema above.

Wheel Variants

Hardware-specific wheels are configured via config/wheel_variants.json:

{
  "packages": {
    "torch": {
      "variants": ["cpu", "cu118", "cu121", "cu124", "rocm5.7"]
    }
  },
  "cuda_mappings": {
    "12.1": "cu121",
    "12.4": "cu124"
  },
  "rocm_variant": "rocm5.7"
}

Override the file path or selection strategy if needed:

config :snakebridge,
  wheel_config_path: "config/wheel_variants.json",
  wheel_strategy: SnakeBridge.WheelSelector.ConfigStrategy

Configuration

Python Dependencies (mix.exs)

def project do
  [
    app: :my_app,
    deps: deps(),
    python_deps: python_deps()
  ]
end

defp python_deps do
  [
    # Simple: name and version
    {:numpy, "1.26.0"},

    # With options (3-tuple)
    {:pandas, "2.0.0",
      pypi_package: "pandas",
      extras: ["sql", "excel"],      # pip extras
      include: ["DataFrame", "read_csv", "read_json"],
      exclude: ["testing"],
      streaming: ["read_csv_chunked"],
      submodules: true},

    # Standard library (no version needed)
    {:json, :stdlib},
    {:math, :stdlib}
  ]
end

Application Config (config/config.exs)

config :snakebridge,
  # Paths
  generated_dir: "lib/snakebridge_generated",
  metadata_dir: ".snakebridge",
  scan_paths: ["lib"],
  scan_exclude: ["lib/generated"],

  # Behavior
  auto_install: :dev_test, # :never | :dev | :dev_test | :always
  strict: false,           # or SNAKEBRIDGE_STRICT=1
  verbose: false,
  error_mode: :raw,        # :raw | :translated | :raise_translated
  atom_allowlist: ["ok", "error"]

# Advanced introspection config
config :snakebridge, :introspector,
  max_concurrency: 4,
  timeout: 30_000

Runtime Config (config/runtime.exs)

SnakeBridge provides a configuration helper that automatically sets up Snakepit with the correct Python executable, adapter, and PYTHONPATH. Add this to your config/runtime.exs:

import Config

# Auto-configure snakepit for snakebridge
SnakeBridge.ConfigHelper.configure_snakepit!()

This replaces ~30 lines of manual configuration and automatically:

• Finds the Python venv (in .venv, snakebridge dep location, or via $SNAKEBRIDGE_VENV)
• Configures the snakebridge adapter
• Sets up PYTHONPATH with snakepit and snakebridge priv directories

For custom pool sizes, affinity defaults, or explicit venv paths:

SnakeBridge.ConfigHelper.configure_snakepit!(
  pool_size: 4,
  affinity: :strict_queue,
  venv_path: "/path/to/venv"
)

For multi-pool setups with per-pool affinity:

SnakeBridge.ConfigHelper.configure_snakepit!(
  pools: [
    %{name: :hint_pool, pool_size: 2, affinity: :hint},
    %{name: :strict_pool, pool_size: 2, affinity: :strict_queue}
  ]
)

Select a pool per call with pool_name in __runtime__:

SnakeBridge.call("math", "sqrt", [16], __runtime__: [pool_name: :strict_pool])

Refs retain the originating pool_name when provided, so subsequent get_attr/call_method calls reuse the same pool even if you omit pool_name.

Python adapter ref lifecycle (environment variables):

SNAKEBRIDGE_REF_TTL_SECONDS=3600
SNAKEBRIDGE_REF_MAX=10000
SNAKEBRIDGE_ATOM_CLASS=true

Python adapter protocol compatibility (environment variables):

# Strict by default. Set to 1/true/yes to accept legacy payloads without protocol metadata.
SNAKEBRIDGE_ALLOW_LEGACY_PROTOCOL=0

Protocol checks are strict by default. Ensure callers include protocol_version and min_supported_version in runtime payloads (all SnakeBridge.Runtime helpers do this automatically).

Mix Tasks

mix snakebridge.setup          # Install Python packages
mix snakebridge.setup --check  # Verify packages installed
mix snakebridge.verify         # Verify hardware compatibility
mix snakebridge.verify --strict # Fail on any mismatch

Examples

See the examples/ directory:

# Run all examples
./examples/run_all.sh

# Individual examples
cd examples/wrapper_args_example && mix run -e Demo.run
cd examples/class_constructor_example && mix run -e Demo.run
cd examples/streaming_example && mix run -e Demo.run
cd examples/strict_mode_example && mix run -e Demo.run
cd examples/multi_session_example && mix run -e Demo.run
cd examples/affinity_defaults_example && mix run -e Demo.run
cd examples/universal_ffi_example && mix run -e Demo.run  # Universal FFI showcase

Guides

Comprehensive guides are available in the guides/ directory:

• Getting Started - Installation, setup, and your first SnakeBridge call
• Universal FFI - Runtime API for dynamic Python calls without codegen
• Generated Wrappers - Compile-time wrapper generation and configuration
• Refs and Sessions - Python object lifecycle and session management
• Type System - Wire protocol, tagged types, and serialization
• Streaming - Generators, iterators, and streaming functions
• Error Handling - Exception translation and structured errors
• Telemetry - Observability, metrics, and debugging
• Best Practices - Patterns, anti-patterns, and production tips
• Session Affinity - Routing and affinity modes for stateful workloads

Script Execution

For scripts and Mix tasks, use SnakeBridge.run_as_script/2 for safe defaults:

SnakeBridge.run_as_script(fn ->
  {:ok, result} = SnakeBridge.call("math", "sqrt", [16])
  IO.inspect(result)
end)

Defaults are exit_mode: :auto and stop_mode: :if_started. For embedded usage, use exit_mode: :none and stop_mode: :never to avoid stopping the host VM.

You can also set SNAKEPIT_SCRIPT_EXIT to none|halt|stop|auto when no explicit options are provided. SNAKEPIT_SCRIPT_HALT is legacy and deprecated.

Warning: exit_mode: :halt or :stop terminates the entire VM.

Direct Runtime API

For dynamic calls when module/function names aren't known at compile time:

# Direct call
{:ok, result} = SnakeBridge.call("math", "sqrt", [16])

# Dynamic call without codegen
{:ok, result} = SnakeBridge.Runtime.call_dynamic("math", "sqrt", [16])

# Streaming call
SnakeBridge.stream("llm", "generate", ["prompt"], [], fn chunk -> IO.write(chunk) end)

# Release refs when done
:ok = SnakeBridge.Runtime.release_ref(ref)
:ok = SnakeBridge.Runtime.release_session("session-id")

Architecture

SnakeBridge is a compile-time code generator:

1. Scan: Find calls to configured library modules in your code
2. Introspect: Query Python for function/class signatures
3. Generate: Create Elixir wrapper modules with proper arities
4. Lock: Record environment for reproducibility

Runtime calls delegate to Snakepit.

Cross-Cutting Contract (Snakepit + Snakebridge)

Wire Format for JSON Any Payloads

SnakeBridge uses a custom gRPC Any convention: Any.value contains raw UTF-8 JSON bytes (not protobuf-packed), with type_url set to type.googleapis.com/google.protobuf.StringValue.

Reserved payload fields (present in every call):

Tagged type encoding uses __type__ and __schema__ markers:

Protocol Versioning

Compatibility is enforced per-call (not per-session). Both sides check:

• Caller's protocol_version >= adapter's MIN_SUPPORTED_VERSION
• Caller's min_supported_version <= adapter's PROTOCOL_VERSION

Strict by default. To accept legacy payloads without version fields:

SNAKEBRIDGE_ALLOW_LEGACY_PROTOCOL=1

On mismatch, SnakeBridgeProtocolError includes all four version values for diagnostics.

Timeouts and Profiles

SnakeBridge provides configurable timeout defaults that are safer for ML/LLM workloads.

Per-Call Timeout Override

# Explicit timeout (10 minutes)
Numpy.compute(data, __runtime__: [timeout: 600_000])

# Use a named profile
Transformers.generate(prompt, __runtime__: [timeout_profile: :ml_inference])

# For streaming operations
MyLib.stream_data(args, opts, callback, __runtime__: [stream_timeout: 3_600_000])

Built-in Timeout Profiles

Global Configuration

config :snakebridge,
  runtime: [
    # Default profile for all calls
    timeout_profile: :default,

    # Override defaults
    default_timeout: 120_000,        # 2 minutes
    default_stream_timeout: 1_800_000, # 30 minutes

    # Per-library profile mapping
    library_profiles: %{
      "transformers" => :ml_inference,
      "torch" => :batch_job
    },

    # Custom profiles
    profiles: %{
      default: [timeout: 120_000],
      ml_inference: [timeout: 600_000, stream_timeout: 1_800_000],
      batch_job: [timeout: :infinity, stream_timeout: :infinity]
    }
  ]

Escape Hatch

Any other keys in __runtime__ are forwarded directly to Snakepit:

# Pass-through to Snakepit's advanced options
MyLib.func(args, __runtime__: [timeout: 60_000, pool_name: :my_pool, affinity: :strict_queue])

Operational Defaults

Requirements

• Elixir ~> 1.14
• Python 3.8+
• uv - Fast Python package manager (required by snakepit)
• Snakepit ~> 0.9.2

Installing uv

# macOS/Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

# Or via Homebrew
brew install uv

License

MIT