Skip to content

Quick Start Guide - Python Backend

Installation

uv add bfast-py
# or
pip install bfast-py

Basic Usage

Simple Serialization

import b_fast

# Your data
data = [{"id": i, "name": f"User {i}"} for i in range(1000)]

# Convenient 1-liner (compression enabled by default)
encoded = b_fast.encode(data)
print(f"Size: {len(encoded)} bytes")

# 1-liner deserialization (auto-decompresses if compressed)
decoded = b_fast.decode(encoded)

# Or using the BFast class instance
encoder = b_fast.BFast()
encoded = encoder.encode_packed(data)  # compress=True by default
decoded = encoder.decode_packed(encoded)  # decompress=True by default

Compression

By default, B-FAST automatically applies compression (compress=True), reducing payload size by up to 89%.

To disable compression for micro-payloads (< 1KB) or ultra-low CPU latency: 

# Uncompressed serialization
raw_binary = b_fast.encode(data, compress=False)
# or
raw_binary = encoder.encode_packed(data, compress=False)

B-FAST provides built-in Response and StreamingResponse classes for FastAPI and Starlette.

Standard Response

from fastapi import FastAPI
from pydantic import BaseModel
from b_fast import BFastResponse

app = FastAPI()

class User(BaseModel):
    id: int
    name: str
    email: str

@app.get("/users", response_class=BFastResponse)
async def get_users():
    return [User(id=i, name=f"User {i}", email=f"user{i}@example.com") for i in range(1000)]

Streamable HTTP Response (Continuous Streaming) 🌊

Stream binary frames in real-time over HTTP/1.1 (Chunked), HTTP/2, or HTTP/3:

import asyncio
from fastapi import FastAPI
from b_fast import BFastStreamingResponse

app = FastAPI()

@app.get("/stream-users", response_class=BFastStreamingResponse)
async def stream_users():
    async def user_generator():
        for i in range(100):
            yield {"id": i, "name": f"User {i}", "status": "active"}
            await asyncio.sleep(0.05)

    return user_generator()

FastMCP 2.0 Integration 🤖

Transmit large masses of AI tool output using B-FAST decorators to save up to 85% context tokens:

from b_fast.fastmcp import FastMCPBFast, bfast_tool

mcp = FastMCPBFast("Analytics Server")

@mcp.tool()
@bfast_tool(compress=True)
def query_large_dataset(limit: int = 1000) -> list[dict]:
    return [{"id": i, "value": i * 1.5} for i in range(limit)]

Next Steps