Complete, RFC-compliant protocol clients built with AnyIO structured concurrency
AnyRFC provides RFC-compliant protocol clients that prioritize correctness, security, and modern async patterns. Built exclusively with AnyIO for structured concurrency.
🎯 RFC Compliance First - Every implementation passes comprehensive RFC test suites
⚡ Modern Async - Structured concurrency with AnyIO (no asyncio dependency hell)
🔒 Security by Default - TLS everywhere, proper certificate validation, secure authentication
🧪 Battle-Tested - Real-world interoperability testing against major servers
📝 Type Safe - Full mypy compliance with strict typing
🚀 Complete - Full implementations, not toys or demos
pip install anyrfc
import anyio
from anyrfc import WebSocketClient
async def main():
async with WebSocketClient("wss://echo.websocket.org/") as ws:
await ws.send_text("Hello, AnyRFC!")
async for message in ws.receive():
print(f"Received: {message}")
break # Just get the first message
anyio.run(main)
Complete WebSocket implementation with all RFC 6455 features:
from anyrfc import WebSocketClient, CloseCode
async with WebSocketClient("wss://api.example.com/ws") as ws:
# Send different message types
await ws.send_text("Hello!")
await ws.send_binary(b"\\x00\\x01\\x02\\x03")
# Handle incoming messages
async for message in ws.receive():
if isinstance(message, str):
print(f"Text: {message}")
else:
print(f"Binary: {message.hex()}")
if should_close:
await ws.close(CloseCode.NORMAL_CLOSURE)
break
Features:
Battle-tested email clients with full RFC compliance and real-world Gmail compatibility:
from anyrfc import IMAPClient, SMTPClient
# IMAP - Complete email operations
async with IMAPClient("imap.gmail.com", use_tls=True) as imap:
await imap.authenticate({"username": "user", "password": "app_password"})
await imap.select_mailbox("INBOX")
# Search and read emails
messages = await imap.search_messages("UNSEEN")
for msg_id in messages[:5]:
email = await imap.fetch_messages(str(msg_id), "BODY[]")
# Mark as read
await imap.store_message_flags(str(msg_id), [b"\\Seen"], "FLAGS")
# Create drafts with proper literal continuation
await imap.append_message("Drafts", email_content, [b"\\Draft"])
# Extract attachments as binary BLOBs
bodystructure = await imap.fetch_messages(str(msg_id), "BODYSTRUCTURE")
# Parse structure and fetch binary parts...
# SMTP - Send emails with authentication
async with SMTPClient("smtp.gmail.com", use_starttls=True) as smtp:
await smtp.authenticate({"username": "user", "password": "app_password"})
await smtp.send_message(
from_addr="sender@example.com",
to_addrs=["recipient@example.com"],
message="""Subject: Hello from AnyRFC!
This email was sent using AnyRFC's SMTP client!
"""
)
IMAP Features (RFC 9051 Compliant):
Every I/O operation uses AnyIO's structured concurrency primitives:
async def websocket_with_timeout():
async with anyio.create_task_group() as tg:
# Connection with automatic cleanup
tg.start_soon(websocket_handler)
# Heartbeat with cancellation scope
with anyio.move_on_after(30):
tg.start_soon(heartbeat_sender)
from anyrfc.websocket import WebSocketClient
client = WebSocketClient("wss://example.com")
compliance_report = await client.validate_compliance()
# Returns detailed RFC 6455 test results
assert compliance_report["handshake_validation"] == True
assert compliance_report["frame_parsing"] == True
assert compliance_report["close_sequence"] == True
from anyrfc import WebSocketClient
from anyrfc.websocket import WSFrame, OpCode
# Fully typed interfaces
client: WebSocketClient = WebSocketClient("wss://api.example.com")
frame: WSFrame = WSFrame(fin=True, opcode=OpCode.TEXT, payload=b"test")
# MyPy validates everything
reveal_type(client.websocket_state) # WSState
reveal_type(await client.receive()) # Union[str, bytes]
pip install anyrfc
git clone https://github.com/elgertam/anyrfc.git
cd anyrfc
# Install with uv (recommended)
uv sync --all-extras
# Or with pip
pip install -e ".[dev]"
anyio>=4.0.0httpx>=0.25.0 (approved dependency)typing-extensions>=4.0.0from anyrfc import WebSocketClient
import json
async def crypto_prices():
# Binance public data stream (no authentication required)
uri = "wss://data-stream.binance.vision/ws/btcusdt@ticker"
# Use relaxed validation for real-world servers
async with WebSocketClient(uri, strict_rfc_validation=False) as ws:
async for message in ws.receive():
data = json.loads(message)
if 'c' in data: # Current price
price = float(data['c'])
change = float(data['P']) # 24hr change %
print(f"💰 BTC-USDT: ${price:,.2f} ({change:+.2f}%)")
from anyrfc import IMAPClient
import anyio
import re
async def email_monitor():
"""Real-time email monitoring with secret code extraction."""
async with IMAPClient("imap.gmail.com", use_tls=True) as imap:
await imap.authenticate({"username": "user", "password": "app_password"})
await imap.select_mailbox("INBOX")
while True:
# Check for new emails every 5 seconds (production-tested)
unread = await imap.search_messages("UNSEEN")
if unread:
print(f"📧 {len(unread)} new emails!")
for msg_id in unread:
# Fetch email content
email_data = await imap.fetch_messages(str(msg_id), "BODY[]")
email_text = email_data[str(msg_id)][b"BODY[]"].decode()
# Extract verification codes (6 digits)
codes = re.findall(r'\b\d{6}\b', email_text)
if codes:
print(f"🔐 Verification code found: {codes[0]}")
# Mark as read
await imap.store_message_flags(str(msg_id), [b"\\Seen"], "FLAGS")
await anyio.sleep(5) # 5-second polling proven effective
# Run all tests
uv run pytest
# RFC compliance tests
uv run pytest tests/rfc_compliance/ -v
# Real-server interoperability
uv run pytest tests/interop/ -v
# Type checking
uv run mypy src/
# Linting
uv run ruff check src/
AnyRFC is extensively tested against production servers:
AnyRFC is built for high-performance workloads:
# Concurrent WebSocket connections
async def stress_test():
async with anyio.create_task_group() as tg:
for i in range(100):
tg.start_soon(websocket_worker, f"wss://api{i}.example.com")
# Memory-efficient message streaming
async def large_mailbox():
async with IMAPClient("imap.example.com") as imap:
# Stream large mailboxes without loading everything into memory
async for message in imap.fetch_messages("1:*", "BODY[]"):
await process_message(message) # Process one at a time
We welcome contributions! AnyRFC follows strict quality standards:
See CONTRIBUTING.md for detailed guidelines.
Report security issues to: andrew@elgert.org
MIT License - see LICENSE for details.
Any + RFC = Protocol clients that work with any server implementing the RFC standard. Built on AnyIO for structured concurrency.
Built by Andrew M. Elgert • Documentation • Issues • PyPI
FAQs
Complete, RFC-compliant protocol clients using AnyIO structured concurrency
We found that anyrfc demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
/Research
Malicious Nx npm versions stole secrets and wallet info using AI CLI tools; Socket’s AI scanner detected the supply chain attack and flagged the malware.
Security News
CISA’s 2025 draft SBOM guidance adds new fields like hashes, licenses, and tool metadata to make software inventories more actionable.
Security News
A clarification on our recent research investigating 60 malicious Ruby gems.