The SyftBox transport uses the traditional syft_core.Client to provide full RPC, encryption, and event handling capabilities for federated learning.
SyftCoreClient
Adapter that wraps syft_core.Client for use with Syft-Flwr.
Class Definition
from syft_flwr.client import SyftCoreClient
from syft_core import Client
class SyftCoreClient(SyftFlwrClient):
"""Adapter for syft_core.Client - the traditional SyftBox client.
This adapter wraps syft_core.Client and provides full syft-rpc/syft-crypto/syft-event
functionality through get_client().
"""
src/syft_flwr/client/syft_core_client.py:9
Constructor
The underlying SyftBox client instance to wrap
Class Methods
load
@classmethod
def load(cls, filepath: Optional[str] = None) -> SyftCoreClient
filepath
str
default:"~/.syftbox/config.json"
Path to the SyftBox client configuration file
Initialized client instance
from syft_flwr.client import SyftCoreClient
# Load from default location
client = SyftCoreClient.load()
# Load from custom path
client = SyftCoreClient.load("/custom/path/config.json")
print(client.email) # [email protected]
print(client.my_datasite) # Path('/home/user/.syftbox/[email protected]')
Properties
email
@property
def email(self) -> str
config_path
@property
def config_path(self) -> Path
my_datasite
@property
def my_datasite(self) -> Path
datasites
@property
def datasites(self) -> Path
Path to datasites root (e.g., ~/.syftbox/datasites/)
Methods
app_data
def app_data(
self,
app_name: Optional[str] = None,
datasite: Optional[str] = None,
) -> Path
Name of the application (e.g., “flwr/my_fl_app”)
Email address of the datasite owner
Path to app data directory
client = SyftCoreClient.load()
# Get app data for current user
app_path = client.app_data("flwr/diabetes_prediction")
# /home/user/.syftbox/datasites/[email protected]/app_data/flwr/diabetes_prediction
# Get app data for another datasite
other_path = client.app_data("flwr/diabetes_prediction", "[email protected]")
# /home/user/.syftbox/datasites/[email protected]/app_data/flwr/diabetes_prediction
get_client
def get_client(self) -> Client
syft_core.Client for RPC/crypto/event operations.
The wrapped native SyftBox client with full RPC/crypto/event stack
from syft_flwr.client import SyftCoreClient
from syft_core import Client as SyftCoreNativeClient
wrapper = SyftCoreClient.load()
native_client = wrapper.get_client()
# Type check for RPC path selection
if isinstance(native_client, SyftCoreNativeClient):
# Use syft_rpc for message sending
from syft_rpc import rpc
future = rpc.send(
url=rpc.make_url("[email protected]", app_name="flwr/app"),
body=b"message",
client=native_client,
encrypt=True
)
SyftRpc
RPC adapter that wraps syft_rpc for message sending with encryption support.
Class Definition
from syft_flwr.rpc import SyftRpc
from syft_core import Client
class SyftRpc(SyftFlwrRpc):
"""Adapter wrapping syft_rpc for traditional SyftBox.
This adapter provides the full syft_rpc functionality including:
- URL-based message routing
- Futures database for response tracking
- Optional encryption via syft_crypto
"""
src/syft_flwr/rpc/syft_rpc.py:12
Constructor
The SyftBox client for message transport
Name of the FL application (e.g., “flwr/my_app”)
Methods
send
def send(
self,
to_email: str,
app_name: str,
endpoint: str,
body: bytes,
encrypt: bool = False,
) -> str
Recipient’s email address
Name of the FL application
RPC endpoint (e.g., “messages”, “rpc/fit”)
Message body as bytes (serialized Flower message)
Whether to encrypt the message using syft_crypto (X3DH)
Future ID for tracking the response
from syft_flwr.rpc import SyftRpc
from syft_core import Client
client = Client.load()
rpc = SyftRpc(client=client, app_name="flwr/my_app")
# Send encrypted message
future_id = rpc.send(
to_email="[email protected]",
app_name="flwr/my_app",
endpoint="messages",
body=b"serialized_message_data",
encrypt=True
)
print(f"Message sent, future_id: {future_id}")
# Message sent, future_id: 550e8400-e29b-41d4-a716-446655440000
get_response
def get_response(self, future_id: str) -> Optional[bytes]
The future ID returned by send()
Response body as bytes, or None if not ready yet
import time
future_id = rpc.send(
to_email="[email protected]",
app_name="flwr/my_app",
endpoint="messages",
body=b"request_data",
encrypt=True
)
# Poll for response
while True:
response = rpc.get_response(future_id)
if response is not None:
print(f"Got response: {len(response)} bytes")
break
time.sleep(1)
delete_future
def delete_future(self, future_id: str) -> None
The future ID to delete from the futures database
response = rpc.get_response(future_id)
if response is not None:
# Process response
process_message(response)
# Clean up
rpc.delete_future(future_id)
SyftEvents
Event handler that wraps syft_event.SyftEvents for real-time message processing.
Class Definition
from syft_flwr.events import SyftEvents
from syft_core import Client
class SyftEvents(SyftFlwrEvents):
"""Adapter wrapping syft_event.SyftEvents for traditional SyftBox.
This adapter provides the syft-extras stack (syft_rpc/syft_crypto/syft_event)
for FL message handling when running with syft_core.Client.
"""
src/syft_flwr/events/syft_events.py:16
Constructor
Name of the FL application
The SyftBox client instance
How long to keep processed messages before cleanup (e.g., “1d”, “12h”)
How often to run cleanup (e.g., “1d”, “6h”)
Properties
client_email
@property
def client_email(self) -> str
Email address of the current client
app_dir
@property
def app_dir(self) -> Path
Path to the app data directory
is_running
@property
def is_running(self) -> bool
Whether the event loop is currently running
Methods
on_request
def on_request(
self,
endpoint: str,
handler: Callable[[bytes], Optional[Union[str, bytes]]],
auto_decrypt: bool = True,
encrypt_reply: bool = False,
) -> None
The endpoint path (e.g., “/messages”, “/rpc/fit”)
handler
Callable[[bytes], Optional[Union[str, bytes]]]
required
Function that receives message bytes and returns response (str or bytes)
Whether to automatically decrypt incoming encrypted messages
Whether to encrypt the reply message
from syft_flwr.events import SyftEvents
from syft_core import Client
client = Client.load()
events = SyftEvents(
app_name="flwr/my_app",
client=client,
cleanup_expiry="1d",
cleanup_interval="1d"
)
def handle_message(body: bytes) -> bytes:
"""Process incoming FL message and return response."""
print(f"Received message: {len(body)} bytes")
# Deserialize and process message
# ... FL logic here ...
return b"response_data"
# Register handler with encryption
events.on_request(
endpoint="/messages",
handler=handle_message,
auto_decrypt=True,
encrypt_reply=True
)
print("Starting event loop...")
events.run_forever()
run_forever
def run_forever(self) -> None
try:
events.run_forever()
except KeyboardInterrupt:
print("Shutting down...")
events.stop()
stop
Signal the event loop to stop gracefully.
is_cleanup_running
def is_cleanup_running(self) -> bool
Whether the cleanup service is currently running
See Also