SnakeBridge

Type-safe Elixir bindings to Python libraries with compile-time code generation and runtime FFI.

Installation

# mix.exs
def project do
  [
    app: :my_app,
    deps: deps(),
    python_deps: python_deps(),
    compilers: [:snakebridge] ++ Mix.compilers()
  ]
end

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

defp python_deps do
  [
    {:numpy, "1.26.0"},
    {:pandas, "2.0.0", include: ["DataFrame", "read_csv"]},
    {:json, :stdlib}
  ]
end

Add runtime configuration in config/runtime.exs:

import Config
SnakeBridge.ConfigHelper.configure_snakepit!()

Then fetch and compile:

mix deps.get && mix compile
mix snakebridge.setup  # Creates managed venv + installs Python packages

SnakeBridge uses the managed venv at priv/snakepit/python/venv by default; no manual venv setup required.

Quick Start

Universal FFI (Any Python Module)

Call any Python function dynamically without code generation:

# Simple function calls
{:ok, 4.0} = SnakeBridge.call("math", "sqrt", [16])
{:ok, pi} = SnakeBridge.get("math", "pi")

# Create Python objects (refs)
{:ok, path} = SnakeBridge.call("pathlib", "Path", ["/tmp/file.txt"])
{:ok, exists?} = SnakeBridge.method(path, "exists", [])
{:ok, name} = SnakeBridge.attr(path, "name")

# Binary data with explicit bytes encoding
{:ok, md5} = SnakeBridge.call("hashlib", "md5", [SnakeBridge.bytes("data")])
{:ok, hex} = SnakeBridge.method(md5, "hexdigest", [])

# Bang variants raise on error
result = SnakeBridge.call!("json", "dumps", [%{key: "value"}])

Generated Wrappers (Configured Libraries)

Libraries in python_deps get Elixir modules with type hints and docs:

# Call like native Elixir
{:ok, result} = Numpy.mean([1, 2, 3, 4])
{:ok, result} = Numpy.mean([[1, 2], [3, 4]], axis: 0)

# Classes generate new/N constructors
{:ok, df} = Pandas.DataFrame.new(%{"a" => [1, 2], "b" => [3, 4]})

# Discovery APIs
Numpy.__functions__()        # List all functions
Numpy.__search__("mean")     # Search by name

When to Use Which

Both approaches coexist in the same project.

Core Concepts

Python Object References

Non-serializable Python objects return as refs - handles to objects in Python memory:

{:ok, ref} = SnakeBridge.call("collections", "Counter", [["a", "b", "a"]])
SnakeBridge.ref?(ref)  # true

# Call methods and access attributes
{:ok, count} = SnakeBridge.method(ref, "most_common", [2])

Session Management

Refs are scoped to sessions. By default, each Elixir process gets an auto-session:

# Explicit session scope
SnakeBridge.SessionContext.with_session(session_id: "my-session", fn ->
  {:ok, ref} = SnakeBridge.call("pathlib", "Path", ["."])
  # ref.session_id == "my-session"
end)

# Release session explicitly
SnakeBridge.release_auto_session()

Graceful Serialization

Containers preserve structure - only non-serializable leaves become refs:

{:ok, result} = SnakeBridge.call("module", "get_mixed_data", [])
# result = %{"name" => "test", "handler" => %SnakeBridge.Ref{...}}
# String fields accessible directly, handler is a ref

Type Encoding

See Type System Guide for complete mapping.

Configuration

python_deps Options

defp python_deps do
  [
    {:numpy, "1.26.0",
      pypi_package: "numpy",          # PyPI name if different
      extras: ["sql"],                # pip extras
      include: ["array", "mean"],     # Only these symbols
      exclude: ["testing"],           # Exclude these
      module_mode: :public,           # Module discovery mode (see below)
      module_depth: 2,                # Limit submodule depth
      module_include: ["linalg"],     # Force-include specific submodules
      module_exclude: ["testing.*"],  # Exclude submodule patterns
      generate: :all,                 # Generate all symbols
      streaming: ["generate"],        # *_stream variants
      min_signature_tier: :stub},     # Signature quality threshold

    {:math, :stdlib}                  # Standard library module
  ]
end

Module discovery modes (for generate: :all):

• :root / :light - Root module only
• :exports / :api - Root __all__ exported submodules (no package walk)
• :public / :standard - Submodules with public APIs (__all__ or top-level defs)
• :explicit - Only modules/packages that define __all__
• :docs - Docs-defined surface from a manifest file
• :all / :nuclear - All submodules including private

See Generated Wrappers for docs manifest workflow and class method guardrails.

Application Config

# config/config.exs
config :snakebridge,
  generated_dir: "lib/snakebridge_generated",
  generated_layout: :split,  # :split (default) | :single
  metadata_dir: ".snakebridge",
  strict: false,
  error_mode: :raw,  # :raw | :translated | :raise_translated
  atom_allowlist: ["ok", "error"],
  scan_extensions: [".ex", ".exs"]  # Include .exs for script/example scanning

Generated files mirror Python module structure (examplelib/predict/__init__.ex for Examplelib.Predict). See Generated Wrappers for details.

Runtime Pool Config

# config/runtime.exs
SnakeBridge.ConfigHelper.configure_snakepit!(
  pool_size: 4,
  affinity: :strict_queue,
  adapter_env: %{                      # Environment for Python adapter
    "HF_HOME" => "/var/lib/huggingface",
    "CUDA_VISIBLE_DEVICES" => "0"
  }
)

For multi-pool setups, per-pool adapter_env overrides global values. See Configuration.

Runtime Options

Pass via __runtime__: key:

SnakeBridge.call("module", "fn", [args],
  __runtime__: [
    session_id: "custom",
    timeout: 60_000,
    affinity: :strict_queue,
    pool_name: :gpu_pool
  ]
)

Runtime Defaults (Process-Scoped)

Set defaults once per process:

SnakeBridge.RuntimeContext.put_defaults(
  pool_name: :gpu_pool,
  timeout_profile: :ml_inference
)

Or scope them to a block:

SnakeBridge.with_runtime(pool_name: :gpu_pool, timeout_profile: :ml_inference) do
  {:ok, result} = SnakeBridge.call("module", "fn", [args])
  result
end

Helper shortcuts for common option shapes:

SnakeBridge.call("numpy", "mean", [scores], SnakeBridge.rt(pool_name: :gpu_pool))

SnakeBridge.call("numpy", "mean", [scores],
  SnakeBridge.opts(py: [axis: 0], runtime: [pool_name: :gpu_pool])
)

Testing

Use the built-in ExUnit template for automatic setup/teardown:

defmodule MyApp.SomeFeatureTest do
  use SnakeBridge.TestCase, pool: :example_pool

  test "runs pipeline" do
    {:ok, out} = Examplelib.SomeModule.some_call("x", y: 1)
    assert out != nil
  end
end

Advanced Features

Streaming and Generators

# Generators implement Enumerable
{:ok, counter} = SnakeBridge.call("itertools", "count", [1])
Enum.take(counter, 5)  # [1, 2, 3, 4, 5]

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

ML Error Translation

config :snakebridge, error_mode: :translated

# Python errors become structured Elixir errors
%SnakeBridge.Error.ShapeMismatchError{expected: [3, 4], actual: [4, 3]}
%SnakeBridge.Error.OutOfMemoryError{device: :cuda, requested: 2048}

Protocol Integration

Refs implement Elixir protocols:

{:ok, ref} = SnakeBridge.call("builtins", "range", [0, 5])
inspect(ref)        # Uses __repr__
"Range: #{ref}"     # Uses __str__
Enum.count(ref)     # Uses __len__
Enum.to_list(ref)   # Uses __iter__

Session Affinity

For stateful workloads, ensure refs route to the same worker:

SnakeBridge.ConfigHelper.configure_snakepit!(affinity: :strict_queue)

# Or per-call
SnakeBridge.method(ref, "compute", [], __runtime__: [affinity: :strict_fail_fast])

Modes: :hint (default), :strict_queue, :strict_fail_fast

Supervised Execution (v0.14.0+)

Stream workers, callbacks, and session cleanup run under SnakeBridge.TaskSupervisor:

• Deadlock-free callbacks: Callbacks can invoke other callbacks without blocking
• Reliable cleanup: Session cleanup tasks are supervised with configurable timeout
• Stream timeouts: Configure stream_timeout for stream_dynamic operations

# Configure session cleanup timeout
config :snakebridge, session_cleanup_timeout_ms: 10_000  # 10 seconds

# Per-call stream timeout
MyLib.generate_stream(input, __runtime__: [stream_timeout: 300_000])

Telemetry

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

Events: [:snakebridge, :compile, :*], [:snakebridge, :runtime, :call, :*], [:snakebridge, :session, :cleanup], [:snakebridge, :session, :cleanup, :error]

Mix Tasks

mix snakebridge.setup          # Install Python packages
mix snakebridge.setup --check  # Verify installation
mix snakebridge.verify         # Hardware compatibility check
mix snakebridge.regen          # Force wrapper regeneration
mix snakebridge.regen --clean  # Remove generated artifacts before regeneration

# Docs manifest generation (for module_mode: :docs)
mix snakebridge.docs.manifest --library <pkg> --inventory <objects.inv> --out priv/snakebridge/<pkg>.docs.json
mix snakebridge.plan           # Preview generation size for docs manifests

Script Execution

For scripts and Mix tasks:

SnakeBridge.script do
  {:ok, result} = SnakeBridge.call("math", "sqrt", [16])
  IO.inspect(result)
end

SnakeBridge.run_as_script/2 remains available for custom lifecycle options.

Before/After

Test setup

Before:

setup_all do
  Application.ensure_all_started(:snakebridge)
  SnakeBridge.ConfigHelper.configure_snakepit!()
  :ok
end

setup do
  SnakeBridge.Runtime.clear_auto_session()
  on_exit(fn -> SnakeBridge.release_auto_session() end)
end

After:

defmodule MyApp.SomeFeatureTest do
  use SnakeBridge.TestCase, pool: :demo_pool
end

Pool selection defaults

Before:

SnakeBridge.call("numpy", "mean", [scores],
  __runtime__: [pool_name: :analytics_pool, timeout_profile: :ml_inference]
)

After:

SnakeBridge.with_runtime(pool_name: :analytics_pool, timeout_profile: :ml_inference) do
  SnakeBridge.call("numpy", "mean", [scores])
end

Callbacks

Before:

SnakeBridge.SessionContext.with_session([session_id: "shared"], fn ->
  SnakeBridge.call("module", "fn", [fn x -> x end])
end)

After:

SnakeBridge.call("module", "fn", [fn x -> x end])

Guides

Examples

The examples/ directory contains runnable demonstrations:

./examples/run_all.sh                    # Run all
cd examples/basic && mix run -e Demo.run # Individual

Key examples:

• universal_ffi_example - Complete Universal FFI showcase
• multi_session_example - Concurrent isolated sessions
• streaming_example - Callback-based streaming
• error_showcase - ML error translation
• signature_showcase - Signature model and arities

See Examples Overview for the complete list.

Architecture

SnakeBridge operates in two phases:

Compile-time: Scans your code, introspects Python modules, generates typed Elixir wrappers with proper arities and documentation.

Runtime: Delegates calls to Snakepit, which manages a gRPC-connected Python process pool.

Wire Protocol

Uses JSON-over-gRPC with tagged types (__type__, __schema__) for non-JSON values. Protocol version 1 with strict compatibility checking.

Timeout Profiles

SnakeBridge.call("module", "fn", [], __runtime__: [timeout_profile: :ml_inference])

Requirements

• Elixir ~> 1.14
• Python 3.8+
• uv - Fast Python package manager

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

License

MIT