ecko 0.1.0

A modern Python web framework. Flask's simplicity with 2026 expectations.

Ecko

A fast, minimal & modern Python web framework.

from ecko import Ecko

app = Ecko()

@app.get("/")
def hello():
    return {"message": "Hello, World!"}

ecko run app:app

Why Ecko?

• Simple by default - No boilerplate, no configuration, just write handlers
• Type-safe without ceremony - Automatic validation from type hints
• Sync and async, seamlessly - Write sync code, it runs in a threadpool automatically
• Batteries included - CORS, sessions, OpenAPI docs built-in but optional
• Fast - Built on ASGI with msgspec for JSON (5-10x faster than Pydantic)
• WebSockets - Out of the box support for WebSockets

Installation

pip install ecko

Requires Python 3.12+.

Quick Start

Create a new project

ecko new myproject
cd myproject
pip install -e .
ecko run app:app

Or start from scratch

# app.py
from ecko import Ecko

app = Ecko()

@app.get("/")
def home():
    return {"message": "Hello, World!"}

@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"id": user_id, "name": f"User {user_id}"}

@app.post("/users")
def create_user(data: dict):
    return {"created": data}, 201

ecko run app:app

Open http://127.0.0.1:8000

Features

Route Parameters

Parameters are extracted automatically from type hints:

@app.get("/users/{user_id}")
def get_user(
    user_id: int,              # From path (matches {user_id})
    include_posts: bool = False # From query string
):
    return {"id": user_id, "include_posts": include_posts}

GET /users/42?include_posts=true
→ {"id": 42, "include_posts": true}

Type Coercion

Ecko automatically converts strings to the declared type:

@app.get("/search")
def search(
    q: str,                    # Required string
    page: int = 1,             # Optional int with default
    limit: int | None = None,  # Optional, None if missing
    tags: list[str] = [],      # List from ?tags=a&tags=b
):
    return {"query": q, "page": page, "limit": limit, "tags": tags}

Request Body

For POST/PUT/PATCH, dict or msgspec Struct parameters are parsed from JSON:

@app.post("/users")
def create_user(data: dict):
    return {"created": data}, 201

For typed validation, use msgspec Structs:

import msgspec

class CreateUser(msgspec.Struct):
    name: str
    email: str
    age: int | None = None

@app.post("/users")
def create_user(user: CreateUser):
    return {"name": user.name, "email": user.email}

Invalid JSON returns a 422 error automatically.

Explicit Parameter Markers

For edge cases, use explicit markers:

from ecko import Query, Path, Header, Body

@app.get("/items/{item_id}")
def get_item(
    item_id: int = Path(description="The item ID"),
    q: str = Query("", description="Search query"),
    token: str = Header(alias="x-api-token"),
):
    return {"item_id": item_id, "query": q}

Responses

Return values are automatically converted to JSON responses:

# Dict → 200 JSON
return {"data": "value"}

# Tuple → status code
return {"created": True}, 201

# Tuple with headers
return {"data": "value"}, 200, {"X-Custom": "header"}

# Explicit response
from ecko import Response, JSONResponse, HTMLResponse, RedirectResponse

return HTMLResponse("<h1>Hello</h1>")
return RedirectResponse("/login")

Async and Sync Handlers

Both work seamlessly:

@app.get("/sync")
def sync_handler():
    # Runs in threadpool automatically
    time.sleep(1)
    return {"sync": True}

@app.get("/async")
async def async_handler():
    # Native async
    await asyncio.sleep(1)
    return {"async": True}

Request Context

Access request data from anywhere without passing it around:

from ecko import context

@app.before_request
def load_user():
    token = context.request.headers.get("authorization")
    context.user = get_user_from_token(token)

@app.get("/profile")
def profile():
    return {"name": context.user.name}

Before/After Request Hooks

@app.before_request
def authenticate():
    if not is_valid_token(context.request.headers.get("authorization")):
        return {"error": "Unauthorized"}, 401  # Short-circuit

@app.after_request
def add_headers(response):
    response._headers["X-Request-ID"] = generate_id()
    return response

Exception Handling

Built-in HTTP exceptions:

from ecko import NotFound, BadRequest, Unauthorized

@app.get("/users/{user_id}")
def get_user(user_id: int):
    user = db.get_user(user_id)
    if not user:
        raise NotFound("User not found")
    return user

Custom exception handlers:

class RateLimitExceeded(Exception):
    pass

@app.exception_handler(RateLimitExceeded)
def handle_rate_limit(exc):
    return {"error": "Too many requests"}, 429

@app.exception_handler(ValueError)
def handle_value_error(exc):
    return {"error": str(exc)}, 400

Middleware

async def timing_middleware(request, call_next):
    start = time.time()
    response = await call_next()
    response._headers["X-Response-Time"] = f"{time.time() - start:.3f}s"
    return response

app.use(timing_middleware)

CORS

from ecko.middleware import cors

# Allow all origins (development)
app.use(cors())

# Specific origins (production)
app.use(cors(
    origins=["https://myapp.com"],
    methods=["GET", "POST", "PUT", "DELETE"],
    allow_credentials=True,
))

Sessions

Signed cookie-based sessions:

from ecko import context
from ecko.middleware import sessions

app.use(sessions(secret="your-secret-key-min-32-chars!!"))

@app.post("/login")
def login(data: dict):
    context.session["user_id"] = data["user_id"]
    return {"logged_in": True}

@app.get("/profile")
def profile():
    user_id = context.session.get("user_id")
    if not user_id:
        return {"error": "Not logged in"}, 401
    return {"user_id": user_id}

@app.post("/logout")
def logout():
    context.session.clear()
    return {"logged_out": True}

WebSockets

Full WebSocket support with a clean async API:

from ecko import Ecko, WebSocket

app = Ecko()

@app.websocket("/ws")
async def websocket_handler(ws: WebSocket):
    await ws.accept()
    async for message in ws:
        await ws.send_text(f"Echo: {message}")

WebSocket Methods

@app.websocket("/chat/{room_id}")
async def chat(ws: WebSocket):
    # Access path parameters
    room_id = ws.path_params["room_id"]

    # Accept the connection
    await ws.accept()

    # Receive messages
    text = await ws.receive_text()
    data = await ws.receive_json()
    binary = await ws.receive_bytes()

    # Send messages
    await ws.send_text("Hello!")
    await ws.send_json({"status": "ok"})
    await ws.send_bytes(b"\x00\x01")

    # Close the connection
    await ws.close(code=1000, reason="Goodbye")

Handling Disconnections

from ecko import WebSocket, WebSocketDisconnect

@app.websocket("/ws")
async def handler(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            message = await ws.receive_text()
            await ws.send_text(f"Got: {message}")
    except WebSocketDisconnect:
        print("Client disconnected")

Or use the async iterator which handles disconnections automatically:

@app.websocket("/ws")
async def handler(ws: WebSocket):
    await ws.accept()
    async for message in ws:
        await ws.send_text(f"Got: {message}")
    # Loop exits cleanly on disconnect

Lifecycle Events

@app.on_startup
async def startup():
    await database.connect()
    print("App started!")

@app.on_shutdown
async def shutdown():
    await database.disconnect()
    print("App stopped!")

OpenAPI Documentation

Auto-generated API docs with Swagger UI:

from ecko import Ecko, setup_openapi

app = Ecko()

setup_openapi(
    app,
    title="My API",
    version="1.0.0",
    description="My awesome API",
)

@app.get("/users/{user_id}")
def get_user(user_id: int):
    """Get a user by their ID."""
    return {"id": user_id}

• GET /docs → Swagger UI
• GET /openapi.json → OpenAPI 3.1 schema

Docstrings become descriptions. Type hints become schemas.

CLI

# Run with auto-reload (development)
ecko run app:app

# Run in production mode
ecko run app:app --prod --workers 4

# Custom host/port
ecko run app:app --host 0.0.0.0 --port 3000

# Create new project
ecko new myproject

# List all routes
ecko routes app:app

Full Example

"""A complete Ecko application."""

import msgspec
from ecko import Ecko, Query, Header, context, NotFound, setup_openapi
from ecko.middleware import cors, sessions


class CreateTodo(msgspec.Struct):
    title: str
    completed: bool = False


app = Ecko()

# Documentation
setup_openapi(app, title="Todo API", version="1.0.0")

# Middleware
app.use(cors())
app.use(sessions(secret="change-me-in-production-use-env-var"))

# In-memory storage (use a real database in production)
todos: dict[int, dict] = {}
next_id = 1


@app.on_startup
def startup():
    print("Todo API ready!")


@app.before_request
def request_logging():
    print(f"{context.request.method} {context.request.path}")


@app.get("/")
def home():
    """API information."""
    return {
        "name": "Todo API",
        "docs": "/docs",
    }


@app.get("/todos")
def list_todos(completed: bool | None = None, limit: int = Query(10)):
    """List all todos with optional filtering."""
    result = list(todos.values())
    if completed is not None:
        result = [t for t in result if t["completed"] == completed]
    return {"todos": result[:limit]}


@app.get("/todos/{todo_id}")
def get_todo(todo_id: int):
    """Get a specific todo by ID."""
    if todo_id not in todos:
        raise NotFound(f"Todo {todo_id} not found")
    return todos[todo_id]


@app.post("/todos")
def create_todo(todo: CreateTodo):
    """Create a new todo."""
    global next_id
    new_todo = {
        "id": next_id,
        "title": todo.title,
        "completed": todo.completed,
    }
    todos[next_id] = new_todo
    next_id += 1
    return new_todo, 201


@app.put("/todos/{todo_id}")
def update_todo(todo_id: int, data: dict):
    """Update a todo."""
    if todo_id not in todos:
        raise NotFound(f"Todo {todo_id} not found")
    todos[todo_id].update(data)
    return todos[todo_id]


@app.delete("/todos/{todo_id}")
def delete_todo(todo_id: int):
    """Delete a todo."""
    if todo_id not in todos:
        raise NotFound(f"Todo {todo_id} not found")
    del todos[todo_id]
    return {"deleted": True}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

API Reference

Core

Export	Description
Ecko	Main application class
Request	Request object
WebSocket	WebSocket connection
WebSocketDisconnect	Exception for disconnections
context	Request-scoped context
setup_openapi	Enable OpenAPI docs

Parameter Markers

Export	Description
Query	Explicit query parameter
Path	Explicit path parameter
Header	Extract from headers
Body	Explicit body parameter

Responses

Export	Description
Response	Base response
JSONResponse	JSON response
HTMLResponse	HTML response
RedirectResponse	HTTP redirect

Exceptions

Export	Description
HTTPException	Base HTTP exception
BadRequest	400 error
Unauthorized	401 error
Forbidden	403 error
NotFound	404 error
MethodNotAllowed	405 error
ValidationError	422 error
InternalServerError	500 error

Middleware

from ecko.middleware import cors, sessions

Export	Description
cors	CORS middleware factory
sessions	Session middleware factory

Comparison

Feature	Ecko	Flask	FastAPI
Async support	Native	Extension	Native
WebSockets	Built-in	Extension	Built-in
Type validation	Auto	Manual	Pydantic
OpenAPI docs	Built-in	Extension	Built-in
Dependency injection	Context	g object	Depends()
Learning curve	Low	Low	Medium
Performance	Fast	Moderate	Fast

TODO

• Default HTML page for new projects
• More examples
• Styled exception pages with comprehensive stack traces
• More middleware
• Native OAuth support
• Robust configuration pattern
• Project organization and structure
• Enriched CLI tooling for easier project management and scaffolding
• More tests
• More docs
• More examples
• Built-in authentication framework
• Built-in event bus
• Built-in task queue
• Built-in caching
• Built-in file storage interface compatible with block storage services
• Pre-built integrations with popular services
• Pre-configured CI/CD pipeline scripts with Github Actions
• Pre-configured Docker images
• Pre-configured logging and monitoring scaffolds for AWS and GCP
• start.ecko.sh page to get started quickly

License

MIT

Links

• Author: https://github.com/sn
• Website: https://ecko.sh
• GitHub: https://github.com/ecko-sh/ecko