# Streaming
Snakepit supports streaming execution for real-time progress updates, large data transfers, and continuous data streams.
## When to Use Streaming
- **Progress updates**: Long-running ML operations reporting incremental progress
- **Large data transfers**: Datasets too large for single responses
- **Real-time data**: Continuous sensor data, log streams, or live predictions
- **Batch processing**: Results delivered as items complete
For simple request-response patterns, use `Snakepit.execute/3` instead.
## Snakepit.execute_stream/4
```elixir
@spec execute_stream(command(), args(), callback_fn(), keyword()) ::
:ok | {:error, Snakepit.Error.t()}
```
| Parameter | Type | Description |
|-----------|------|-------------|
| `command` | `String.t()` | The streaming command |
| `args` | `map()` | Command arguments |
| `callback_fn` | `(term() -> any())` | Called for each chunk |
| `opts` | `keyword()` | Execution options |
| Option | Default | Description |
|--------|---------|-------------|
| `:pool` | `Snakepit.Pool` | Pool to use |
| `:timeout` | `300000` | Timeout in ms |
| `:session_id` | `nil` | Session affinity |
### Basic Example
```elixir
:ok = Snakepit.execute_stream("process_items", %{items: list}, fn chunk ->
IO.puts("Received: #{inspect(chunk)}")
end)
```
## Callback Function
The callback is invoked for each chunk, synchronously and in order.
```elixir
# Logging
callback = fn chunk -> IO.inspect(chunk) end
# Sending to another process
callback = fn chunk -> send(consumer_pid, {:chunk, chunk}) end
```
## Python Streaming Tools
### Enabling Streaming
Use `supports_streaming=True` in the `@tool` decorator:
```python
from snakepit_bridge.base_adapter import BaseAdapter, tool
class MyAdapter(BaseAdapter):
@tool(description="Stream progress", supports_streaming=True)
def stream_progress(self, steps: int = 10):
for i in range(steps):
yield {"step": i + 1, "total": steps, "progress": (i + 1) / steps * 100}
```
### Using Generators
Each `yield` becomes a chunk sent to Elixir:
```python
@tool(description="Stream numbers", supports_streaming=True)
def stream_numbers(self, count: int = 10):
for i in range(count):
yield {"number": i, "is_last": i == count - 1}
```
### StreamChunk Class
For explicit control over the final marker:
```python
from snakepit_bridge.adapters.showcase.tool import StreamChunk
@tool(description="Stream with metadata", supports_streaming=True)
def stream_items(self, count: int = 5):
for i in range(count):
yield StreamChunk(
data={"index": i + 1, "value": f"Item {i + 1}"},
is_final=(i == count - 1)
)
```
## Progress Update Patterns
### Training Progress
```python
@tool(description="Train with progress", supports_streaming=True)
def train_model(self, epochs: int = 10):
for epoch in range(epochs):
loss = 1.0 / (epoch + 1)
accuracy = min(0.95, 0.5 + epoch * 0.05)
yield {"type": "progress", "epoch": epoch + 1, "loss": loss, "accuracy": accuracy}
yield {"type": "complete", "final_loss": loss, "final_accuracy": accuracy}
```
Elixir consumer:
```elixir
:ok = Snakepit.execute_stream("train_model", %{epochs: 10}, fn chunk ->
case chunk["type"] do
"progress" -> IO.puts("Epoch #{chunk["epoch"]}: loss=#{chunk["loss"]}")
"complete" -> IO.puts("Training complete!")
end
end)
```
## Large Data Transfer Patterns
### Chunked Dataset
```python
@tool(description="Generate dataset", supports_streaming=True)
def generate_dataset(self, rows: int = 10000, chunk_size: int = 1000):
import numpy as np
total_sent = 0
while total_sent < rows:
batch = min(chunk_size, rows - total_sent)
data = np.random.randn(batch, 10)
total_sent += batch
yield StreamChunk(
data={"rows": batch, "total": total_sent, "data": data.tolist()},
is_final=(total_sent >= rows)
)
```
## Error Handling in Streams
### Python Errors
Exceptions propagate to Elixir:
```elixir
case Snakepit.execute_stream("may_fail", %{}, callback) do
:ok -> IO.puts("Success")
{:error, error} -> IO.puts("Failed: #{error.message}")
end
```
### Stream Cancellation
```elixir
task = Task.async(fn ->
Snakepit.execute_stream("infinite_stream", %{}, fn chunk ->
IO.puts("Received: #{inspect(chunk)}")
end)
end)
Process.sleep(5000)
Task.shutdown(task, :brutal_kill) # Cancel stream
```
## Complete Streaming Example
### Python Adapter
```python
from snakepit_bridge.base_adapter import BaseAdapter, tool
from snakepit_bridge.adapters.showcase.tool import StreamChunk
import time
class MLStreamingAdapter(BaseAdapter):
@tool(description="Train with streaming progress", supports_streaming=True)
def train_with_progress(self, epochs: int = 10):
loss, accuracy = 1.0, 0.5
for epoch in range(epochs):
time.sleep(0.5)
loss *= 0.8
accuracy = min(0.99, accuracy + 0.05)
yield StreamChunk(
data={"epoch": epoch + 1, "loss": round(loss, 4), "accuracy": round(accuracy, 4)},
is_final=False
)
yield StreamChunk(
data={"status": "complete", "model_id": f"model_{int(time.time())}"},
is_final=True
)
```
### Elixir Consumer
```elixir
defmodule MyApp.MLClient do
require Logger
def train_model(epochs) do
Logger.info("Starting training")
result = Snakepit.execute_stream("train_with_progress", %{epochs: epochs}, fn chunk ->
if chunk["status"] == "complete" do
Logger.info("Complete! Model: #{chunk["model_id"]}")
else
Logger.info("Epoch #{chunk["epoch"]}: loss=#{chunk["loss"]}, acc=#{chunk["accuracy"]}")
end
end)
case result do
:ok -> {:ok, "Training completed"}
{:error, e} -> {:error, e}
end
end
end
```
## Performance Considerations
1. **Chunk size**: Balance responsiveness vs efficiency (1000-10000 items typical)
2. **Callback overhead**: Keep callbacks lightweight; offload heavy processing
3. **Timeouts**: Default is 5 minutes; adjust with `:timeout` option
4. **Memory**: Streaming reduces peak memory by processing incrementally
5. **gRPC required**: Streaming only works with gRPC adapters
## Server-Side Streaming Implementation (v0.8.5+)
Starting in v0.8.5, Snakepit's BridgeServer fully implements `ExecuteStreamingTool`,
enabling end-to-end gRPC streaming from external clients through to Python workers.
### Requirements for Streaming Tools
1. Tool must be a **remote** (Python) tool
2. Tool must have `supports_streaming: true` in metadata
3. Python adapter must implement the streaming tool as a generator
### Enabling a Tool for Streaming
In your Python adapter:
```python
@tool(description="Stream results", supports_streaming=True)
def my_streaming_tool(self, param: str):
for i in range(10):
yield {"step": i, "result": f"Processing {param}"}
```
The tool will be registered with streaming support automatically.
### Stream Chunk Metadata
Final chunks include automatic metadata decoration:
- `execution_time_ms`: Total execution time in milliseconds
- `tool_type`: The type of tool (always "remote" for streaming)
- `worker_id`: The ID of the worker that executed the tool
If the worker doesn't send a final chunk, Snakepit injects a synthetic final chunk
with `synthetic_final: "true"` in metadata.
