Skip to main content
Sifr’s networking stack is a native async substrate built on Rust’s async I/O primitives and the Rustls TLS library. It is not a compatibility shim over CPython’s socket, ssl, or http modules — those bare names are rejected with a diagnostic and the correct sifr.* replacement is suggested. This page covers the four public networking modules and how they compose.

​
Module Map

ModulePurpose
sifr.netAsync TCP connections and listeners
sifr.httpHTTP protocol primitives: methods, headers, bodies, status codes
sifr.tlsRustls-backed TLS client configuration
sifr.urlTyped URL parsing, query strings, and percent-encoding
The following CPython-shaped imports are rejected in Sifr:
import socket          # ❌ Use sifr.net
import ssl             # ❌ Use sifr.tls
from urllib.parse import urlparse   # ❌ Use sifr.url
from http.server import HTTPServer  # ❌ Use sifr.http

​
TCP with sifr.net

connect_tcp opens an async TCP stream to a remote address. listen_tcp binds a local port and returns a listener that accepts incoming connections: 
from sifr.net import connect_tcp, listen_tcp

# Connect to a remote host
async def send_ping(host: str, port: int) -> None:
    stream = await connect_tcp(host, port)
    await stream.write(b"ping\n")
    response: bytes = await stream.read(1024)
    print(response)
    stream.close()

# Listen for incoming connections
async def echo_server(port: int) -> None:
    listener = await listen_tcp(port)
    while True:
        conn = await listener.accept()
        data: bytes = await conn.read(4096)
        await conn.write(data)
        conn.close()
Network operations consume Sifr’s concurrency runtime for cancellation, deadlines, and backpressure. Use task.scope() or task.timeout() from sifr.task to bound the lifetime of network operations.

​
TLS with sifr.tls

Wrap a TCP connection in TLS using a client configuration built from a CA certificate bundle: 
from sifr.net import connect_tcp
from sifr.tls import client_config_from_ca_pem

async def fetch_tls(host: str, port: int, ca_pem: bytes) -> bytes:
    config = client_config_from_ca_pem(ca_pem)
    stream = await connect_tcp(host, port)
    tls_stream = await config.connect(host, stream)
    await tls_stream.write(b"GET / HTTP/1.1\r\nHost: " + host.encode() + b"\r\n\r\n")
    response: bytes = await tls_stream.read(8192)
    tls_stream.close()
    return response
TLS is backed by Rustls, which enforces modern cipher suites and certificate validation by default. There is no API to disable certificate verification.

​
HTTP Substrate with sifr.http

sifr.http provides the typed HTTP protocol layer: request and response heads, header maps, body streams, and cookie parsing. This is a substrate API — it gives you the building blocks rather than a full client or server framework.

​
Headers

Build a HeaderMap from a list of (name, value) pairs using headers_from_pairs: 
from sifr.http import HeaderMap, headers_from_pairs

headers: HeaderMap = headers_from_pairs([
    ("content-type", "application/json"),
    ("x-request-id", "abc123"),
])

​
Request and Response Heads

Construct typed request and response heads with request_head and response_head: 
from sifr.http import RequestHead, ResponseHead, request_head, response_head, headers_from_pairs

req_headers: HeaderMap = headers_from_pairs([
    ("content-type", "application/octet-stream"),
])
request: RequestHead = request_head("POST", "/api/upload", req_headers)

print(request.method.value)   # POST
print(request.path)           # /api/upload

resp_headers: HeaderMap = headers_from_pairs([("x-sifr", "ok")])
response: ResponseHead = response_head(200, resp_headers, "HTTP/1.1")
print(response.status.code)   # 200

​
Body Streams

BodyStream lets you collect a response body with an explicit size limit, preventing runaway memory growth from oversized payloads: 
from sifr.http import BodyStream, BodyError, body_from_chunks

body: BodyStream = body_from_chunks([b"hello", b" ", b"world"])
collected: bytes = body.collect_with_limit(1024)
print(collected)   # b"hello world"

# Enforce limits on untrusted input
try:
    limited_body: BodyStream = body_from_chunks([b"very long payload..."])
    data: bytes = limited_body.collect_with_limit(8)
except BodyError as e:
    print(e.message)   # "exceeds limit"
Parse the Cookie header into a list of (name, value) tuples: 
from sifr.http import parse_cookie_header

cookies: list[tuple[str, str]] = parse_cookie_header("session=demo; mode=substrate")
# [("session", "demo"), ("mode", "substrate")]

​
Full HTTP Substrate Demo

The following is the complete HTTP substrate demo from the Sifr repository: 
from sifr.http import (
    BodyError,
    BodyStream,
    HeaderMap,
    RequestHead,
    ResponseHead,
    body_from_chunks,
    headers_from_pairs,
    parse_cookie_header,
    request_head,
    response_head,
)


def main() -> Result[None, Error]:
    try:
        headers: HeaderMap = headers_from_pairs([
            ("content-type", "application/octet-stream"),
            ("cookie", "session=demo; mode=substrate"),
        ])
        request: RequestHead = request_head("POST", "/demo/network-http", headers)
        assert request.method.value == "POST"
        assert request.path == "/demo/network-http"

        cookies: list[tuple[str, str]] = parse_cookie_header("session=demo; mode=substrate")
        assert len(cookies) == 2

        body: BodyStream = body_from_chunks([b"ping", b"-http"])
        collected: bytes = body.collect_with_limit(32)
        assert collected == b"ping-http"

        response_headers: HeaderMap = headers_from_pairs([("x-sifr-demo", "network-http")])
        response: ResponseHead = response_head(201, response_headers, "HTTP/1.1")
        assert response.status.code == 201

        too_large: bool = False
        try:
            limited_body: BodyStream = body_from_chunks([b"abcd"])
            _limited: bytes = limited_body.collect_with_limit(2)
        except BodyError as e:
            too_large = "exceeds" in e.message
        assert too_large
    except Error as e:
        raise e
    return None

​
URL Parsing with sifr.url

Parse and decompose URLs with sifr.url: 
from sifr.url import parse_url

url = parse_url("https://example.com/search?q=sifr&page=2")

print(url.scheme())    # https
print(url.host())      # example.com
print(url.path())      # /search
print(url.query())     # q=sifr&page=2
Query parameters come back as typed key-value pairs. Percent-encoding and decoding are handled automatically for standard ASCII-safe characters.
Non-ASCII URL encodings (e.g., IDNA hostname canonicalization) are currently blocked on text/i18n Unicode alignment work and are not available in this release.

​
What Is Not in This Release

The networking substrate is production-ready for TCP, TLS, and HTTP/1.1 and HTTP/2 transport. The following features are deferred to later phases:
  • High-level HTTP client (retries, auth, compression, redirects)
  • Server routing, middleware, and request extractors
  • WebSocket and HTTP/3 support
  • SO_REUSEPORT and multi-worker serving