# Bidirectional Tool Bridge - Technical Documentation
## Overview
The Bidirectional Tool Bridge enables seamless cross-language function execution between Elixir and Python in the Snakepit framework. This allows developers to leverage the strengths of both languages within a single application, calling Python functions from Elixir and Elixir functions from Python transparently through gRPC.
## Architecture
### Core Components
```
┌─────────────────────────────────────────────────────────────────┐
│ Elixir Side │
├─────────────────────────────────────────────────────────────────┤
│ ToolRegistry (GenServer) │
│ - ETS table: :snakepit_tool_registry │
│ - Manages tool metadata and execution │
│ │
│ BridgeServer (gRPC handlers) │
│ - register_tools/2 │
│ - get_exposed_elixir_tools/2 │
│ - execute_elixir_tool/2 │
│ - execute_tool/2 │
└─────────────────────────────────────────────────────────────────┘
↕ gRPC
┌─────────────────────────────────────────────────────────────────┐
│ Python Side │
├─────────────────────────────────────────────────────────────────┤
│ SessionContext │
│ - Auto-discovers Elixir tools on init │
│ - Creates dynamic proxies for natural Python syntax │
│ - Manages tool execution and serialization │
│ │
│ BaseAdapter │
│ - @tool decorator for exposing Python functions │
│ - Automatic tool discovery and registration │
└─────────────────────────────────────────────────────────────────┘
```
## Quick Start
### 1. Expose Elixir Functions to Python
```elixir
# In your Elixir code
alias Snakepit.Bridge.ToolRegistry
# Register a function to be callable from Python
ToolRegistry.register_elixir_tool(
session_id,
"parse_json",
&MyModule.parse_json/1,
%{
description: "Parse a JSON string",
exposed_to_python: true, # Important!
parameters: [
%{name: "json_string", type: "string", required: true}
]
}
)
```
**Note**: As of v0.4.1, the tool bridge includes automatic session initialization and enhanced tool capabilities. Sessions are automatically created when Python tools attempt to register, eliminating the need for manual session setup in most cases.
### New in v0.4.1
- **Enhanced ShowcaseAdapter** with `process_text` and `get_stats` tools
- **Fixed gRPC tool registration** - Resolved async/sync issues with proper response handling
- **Complete remote tool dispatch** - Bidirectional communication between Elixir and Python
### 2. Call Elixir Functions from Python
```python
from snakepit_bridge.session_context import SessionContext
# Connect to session
ctx = SessionContext(stub, session_id)
# Method 1: Direct call
result = ctx.call_elixir_tool("parse_json", json_string='{"test": true}')
# Method 2: Using tool proxy (more Pythonic)
result = ctx.elixir_tools["parse_json"](json_string='{"test": true}')
```
### 3. Expose Python Functions to Elixir
```python
from snakepit_bridge.base_adapter import BaseAdapter, tool
class MyAdapter(BaseAdapter):
@tool(description="Process data with Python")
def process_data(self, data: str, mode: str = "default") -> dict:
# Your Python logic here
return {"processed": data, "mode": mode}
# Register with session (happens automatically in grpc_server.py)
adapter = MyAdapter()
adapter.register_with_session(session_id, stub)
```
### 4. Call Python Functions from Elixir
```elixir
# Execute a Python tool (once registered)
{:ok, result} = ToolRegistry.execute_tool(session_id, "process_data", %{
"data" => "test input",
"mode" => "advanced"
})
```
## Detailed Implementation Guide
### Setting Up Elixir Tools
1. **Define Your Function**
```elixir
defmodule MyTools do
def calculate_stats(params) do
list = Map.get(params, "numbers", [])
%{
sum: Enum.sum(list),
mean: Enum.sum(list) / length(list),
max: Enum.max(list)
}
end
end
```
2. **Register During Application Startup**
```elixir
# In your application.ex or supervisor
ToolRegistry.register_elixir_tool(
session_id,
"calculate_stats",
&MyTools.calculate_stats/1,
%{
description: "Calculate statistics for a list of numbers",
exposed_to_python: true,
parameters: [
%{name: "numbers", type: "array", required: true}
]
}
)
```
### Setting Up Python Tools
1. **Create an Adapter Class**
```python
from snakepit_bridge.base_adapter import BaseAdapter, tool
import numpy as np
class DataAdapter(BaseAdapter):
@tool(description="Perform FFT on signal data")
def fft_analysis(self, signal: list, sample_rate: int = 1000) -> dict:
fft_result = np.fft.fft(signal)
frequencies = np.fft.fftfreq(len(signal), 1/sample_rate)
return {
"dominant_frequency": float(frequencies[np.argmax(np.abs(fft_result))]),
"magnitude": float(np.max(np.abs(fft_result))),
"sample_rate": sample_rate
}
```
2. **Integration with gRPC Server**
```python
# In your grpc_server.py
adapter = DataAdapter()
# This happens automatically when adapter is used in ExecuteTool
tool_names = adapter.register_with_session(session_id, stub)
```
### Parameter Types and Serialization
The bridge supports these parameter types:
| Type | Elixir | Python | Notes |
|------|--------|---------|-------|
| string | String.t() | str | Direct mapping |
| integer | integer() | int | Direct mapping |
| float | float() | float | Direct mapping |
| boolean | boolean() | bool | Direct mapping |
| array | list() | list | JSON serialization |
| object | map() | dict | JSON serialization |
| null | nil | None | Direct mapping |
**Complex Type Example:**
```python
# Python sending complex data to Elixir
result = ctx.call_elixir_tool("process_data",
matrix=[[1, 2], [3, 4]],
config={"threshold": 0.5, "mode": "fast"}
)
```
### Error Handling
```python
# Python side
try:
result = ctx.call_elixir_tool("risky_operation", data=input_data)
except RuntimeError as e:
print(f"Elixir tool failed: {e}")
# Handle error appropriately
```
```elixir
# Elixir side
def risky_operation(params) do
case validate_input(params) do
:ok ->
{:ok, perform_operation(params)}
{:error, reason} ->
{:error, "Validation failed: #{reason}"}
end
end
```
## Advanced Features
### 1. Session-Scoped Tools
Tools are isolated by session, allowing multi-tenant usage:
```python
# Different sessions have different tools
ctx1 = SessionContext(stub, "session-1")
ctx2 = SessionContext(stub, "session-2")
# Each context only sees tools registered for its session
print(ctx1.elixir_tools.keys()) # Tools for session-1
print(ctx2.elixir_tools.keys()) # Tools for session-2
```
### 2. Dynamic Tool Discovery
Tools are discovered automatically:
```python
# Python discovers Elixir tools on initialization
ctx = SessionContext(stub, session_id)
# ctx.elixir_tools is automatically populated
# List all available Elixir tools
for name, proxy in ctx.elixir_tools.items():
print(f"{name}: {proxy.__doc__}")
```
### 3. Metadata and Introspection
```python
# Access tool metadata
tool_proxy = ctx.elixir_tools["calculate_stats"]
print(tool_proxy.__name__) # "calculate_stats"
print(tool_proxy.__doc__) # Full description with parameters
```
### 4. Hybrid Processing Patterns
```python
# Combine Python and Elixir strengths
class HybridAdapter(BaseAdapter):
def __init__(self, session_context):
self.ctx = session_context
@tool(description="Hybrid ML pipeline")
def ml_pipeline(self, raw_data: list) -> dict:
# Python: Data preprocessing with pandas/numpy
processed = self.preprocess(raw_data)
# Elixir: Parallel processing
distributed_result = self.ctx.call_elixir_tool(
"parallel_map",
data=processed,
function="complex_calculation"
)
# Python: ML model inference
predictions = self.model.predict(distributed_result["results"])
return {
"predictions": predictions.tolist(),
"processing_time": distributed_result["elapsed_ms"]
}
```
## Performance Considerations
### Benchmarks
Typical overhead for cross-language calls:
- Simple parameter types: ~1-2ms
- Complex objects (1KB JSON): ~5-10ms
- Large arrays (10MB): ~50-100ms
### Optimization Tips
1. **Batch Operations**
```python
# Instead of multiple calls
results = []
for item in items:
results.append(ctx.call_elixir_tool("process", data=item))
# Use a single batched call
results = ctx.call_elixir_tool("process_batch", items=items)
```
2. **Cache Frequently Used Tools**
```python
# Store proxy reference
process_fn = ctx.elixir_tools["process_data"]
# Reuse in hot loops
for data in large_dataset:
result = process_fn(data=data)
```
3. **Minimize Serialization**
```elixir
# Return only necessary data
def analyze_large_dataset(params) do
dataset = decode_dataset(params["data"])
# Don't return the entire dataset
%{
summary: calculate_summary(dataset),
metrics: extract_metrics(dataset),
# Not: processed_data: dataset
}
end
```
## Troubleshooting
### Common Issues
1. **"Tool not found" Error**
- Ensure `exposed_to_python: true` is set in Elixir tool metadata
- Verify the session ID matches between Python and Elixir
- Check that the Elixir tool was registered before Python tried to discover it
2. **Serialization Errors**
- Ensure all return values are JSON-serializable
- Use basic types (string, number, boolean, array, object)
- For binary data, base64 encode it first
3. **Connection Refused**
- Verify gRPC server is running: `lsof -i :50051`
- Check firewall settings
- Ensure correct host/port in Python client
4. **Session Already Exists**
```elixir
# Clean up before creating
SessionStore.delete_session(session_id)
{:ok, _} = SessionStore.create_session(session_id, metadata: %{})
```
### Debug Mode
Enable detailed logging:
```python
# Python
import logging
logging.getLogger('snakepit_bridge').setLevel(logging.DEBUG)
```
```elixir
# Elixir - in config.exs
config :logger, level: :debug
```
## Security Considerations
1. **Input Validation**: Always validate parameters in tool implementations
2. **Access Control**: Implement session-based permissions if needed
3. **Rate Limiting**: Consider implementing rate limits for expensive operations
4. **Sanitization**: Sanitize error messages to avoid leaking sensitive information
## Future Enhancements
### Currently In Development
1. **Streaming Support**: For long-running operations
```python
# Future API
for progress in ctx.stream_elixir_tool("long_operation", data=data):
print(f"Progress: {progress.percent}%")
```
2. **Binary Data Optimization**: Direct binary transfer without JSON encoding
3. **Connection Pooling**: Reuse gRPC connections for better performance
4. **Automatic Retries**: With exponential backoff for transient failures
## Examples
Complete working examples are available in the `examples/` directory:
- `bidirectional_tools_demo.exs` - Elixir server with example tools
- `python_elixir_tools_demo.py` - Python client demonstrating all features
- `elixir_python_tools_demo.exs` - Elixir calling Python tools (coming soon)
## Contributing
When adding new features to the tool bridge:
1. Update the protobuf definitions if needed
2. Implement both Elixir and Python sides
3. Add tests for serialization edge cases
4. Update this documentation
5. Add examples demonstrating the feature
## Related Documentation
- [Main README](README.md)
- [Unified gRPC Bridge](README_UNIFIED_GRPC_BRIDGE.md)
- [Testing Guide](README_TESTING.md)
