Python’s wsgiref module is the reference WSGI implementation — a development server, validation middleware, and utility functions for building PEP 3333–compliant WSGI applications. from wsgiref.simple_server import make_server. make_server: srv = wsgiref.simple_server.make_server("", 8000, app) → srv.serve_forever(). WSGI app: a callable with signature (environ, start_response) → iterable of bytes. start_response: start_response("200 OK", [("Content-Type", "text/plain")]). Validator: from wsgiref.validate import validator; wrapped = validator(app) — raises AssertionError on any WSGI spec violation (missing headers, wrong types, etc.). wsgiref.util: wsgiref.util.request_uri(environ) → full request URI; application_uri(environ) → app root; shift_path_info(environ) — pops one path component from PATH_INFO into SCRIPT_NAME. wsgiref.headers: h = wsgiref.headers.Headers([]); h["Content-Type"] = "application/json"; h.add_header("Set-Cookie", "token=abc", path="/") — builds header list for start_response. FileWrapper: wsgiref.util.FileWrapper(fileobj, blksize=8192) — wraps a file object into a lazy iterable for streaming. Handler: wsgiref.handlers.SimpleHandler(stdin, stdout, stderr, environ) — write a WSGI response to arbitrary I/O streams without a real socket. Claude Code generates development servers, WSGI testing harnesses, router middleware, header builder utilities, and zero-dependency API endpoints.
CLAUDE.md for wsgiref
## wsgiref Stack
- Stdlib: from wsgiref.simple_server import make_server
- from wsgiref.validate import validator
- from wsgiref.util import request_uri, shift_path_info, FileWrapper
- from wsgiref.headers import Headers
- App: def app(environ, start_response): start_response("200 OK", headers); yield b"body"
- Dev: with make_server("", 8080, validator(app)) as srv: srv.serve_forever()
- Test: use wsgiref.handlers.SimpleHandler to call app without a real server socketwsgiref WSGI Application Pipeline
# app/wsgiutil.py — request env, router middleware, header builder, test client
from __future__ import annotations
import io
import json
import sys
import threading
import time
import wsgiref.handlers
import wsgiref.headers
import wsgiref.simple_server
import wsgiref.util
import wsgiref.validate
from dataclasses import dataclass, field
from typing import Any, Callable, Iterable, Iterator
# WSGI type aliases
Environ = dict[str, Any]
StartResponse = Callable[[str, list[tuple[str, str]]], Any]
WSGIApp = Callable[[Environ, StartResponse], Iterable[bytes]]
# ─────────────────────────────────────────────────────────────────────────────
# 1. Request helpers
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class Request:
"""
Lightweight wrapper around a WSGI environ dict.
Example:
def app(environ, start_response):
req = Request(environ)
print(req.method, req.path)
"""
environ: Environ
@property
def method(self) -> str:
return self.environ.get("REQUEST_METHOD", "GET").upper()
@property
def path(self) -> str:
return self.environ.get("PATH_INFO", "/")
@property
def query_string(self) -> str:
return self.environ.get("QUERY_STRING", "")
@property
def content_type(self) -> str:
return self.environ.get("CONTENT_TYPE", "")
@property
def content_length(self) -> int:
try:
return int(self.environ.get("CONTENT_LENGTH") or 0)
except ValueError:
return 0
def body(self) -> bytes:
if self.content_length:
return self.environ["wsgi.input"].read(self.content_length)
return b""
def json(self) -> Any:
return json.loads(self.body())
def query_params(self) -> dict[str, str]:
import urllib.parse
return dict(urllib.parse.parse_qsl(self.query_string))
def header(self, name: str) -> str:
key = "HTTP_" + name.upper().replace("-", "_")
return self.environ.get(key, "")
def uri(self) -> str:
return wsgiref.util.request_uri(self.environ)
# ─────────────────────────────────────────────────────────────────────────────
# 2. Response helpers
# ─────────────────────────────────────────────────────────────────────────────
def respond(
start_response: StartResponse,
body: bytes | str,
status: str = "200 OK",
content_type: str = "text/plain; charset=utf-8",
extra_headers: list[tuple[str, str]] | None = None,
) -> list[bytes]:
"""
Build and send a WSGI response. Returns a single-item body list.
Example:
def app(environ, start_response):
return respond(start_response, "Hello!", "200 OK")
"""
if isinstance(body, str):
body = body.encode()
headers = [
("Content-Type", content_type),
("Content-Length", str(len(body))),
]
if extra_headers:
headers.extend(extra_headers)
start_response(status, headers)
return [body]
def respond_json(
start_response: StartResponse,
data: Any,
status: str = "200 OK",
) -> list[bytes]:
"""
Serialize data to JSON and return a WSGI response.
Example:
def api(environ, start_response):
return respond_json(start_response, {"status": "ok"})
"""
body = json.dumps(data).encode()
return respond(start_response, body, status, content_type="application/json")
def respond_redirect(
start_response: StartResponse,
location: str,
permanent: bool = False,
) -> list[bytes]:
"""Send an HTTP redirect response."""
status = "301 Moved Permanently" if permanent else "302 Found"
start_response(status, [("Location", location), ("Content-Length", "0")])
return []
# ─────────────────────────────────────────────────────────────────────────────
# 3. Router middleware
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class Router:
"""
Simple path + method router for WSGI.
Example:
router = Router()
@router.route("/hello", methods=["GET"])
def hello(environ, start_response):
return respond(start_response, "Hello World!")
@router.route("/echo", methods=["POST"])
def echo(environ, start_response):
body = Request(environ).body()
return respond(start_response, body)
run_dev_server(router, port=8080)
"""
_routes: dict[tuple[str, str], WSGIApp] = field(default_factory=dict)
_not_found: WSGIApp | None = None
def route(
self,
path: str,
methods: list[str] | None = None,
) -> Callable[[WSGIApp], WSGIApp]:
def decorator(fn: WSGIApp) -> WSGIApp:
for method in (methods or ["GET"]):
self._routes[(method.upper(), path)] = fn
return fn
return decorator
def not_found_handler(self, fn: WSGIApp) -> WSGIApp:
self._not_found = fn
return fn
def __call__(self, environ: Environ, start_response: StartResponse) -> Iterable[bytes]:
method = environ.get("REQUEST_METHOD", "GET").upper()
path = environ.get("PATH_INFO", "/")
handler = self._routes.get((method, path))
if handler is None:
if self._not_found:
return self._not_found(environ, start_response)
return respond(start_response, f"404 Not Found: {path}",
status="404 Not Found")
return handler(environ, start_response)
# ─────────────────────────────────────────────────────────────────────────────
# 4. Middleware
# ─────────────────────────────────────────────────────────────────────────────
def logging_middleware(app: WSGIApp) -> WSGIApp:
"""
WSGI middleware that logs method, path, and status to stdout.
Example:
logged_app = logging_middleware(router)
"""
def wrapped(environ: Environ, start_response: StartResponse) -> Iterable[bytes]:
method = environ.get("REQUEST_METHOD", "-")
path = environ.get("PATH_INFO", "-")
status_holder: list[str] = []
def capturing_sr(status: str, headers: list, *args) -> Any:
status_holder.append(status)
return start_response(status, headers, *args)
result = app(environ, capturing_sr)
status = status_holder[0] if status_holder else "?"
print(f"{method} {path} → {status}")
return result
return wrapped
def cors_middleware(app: WSGIApp, allowed_origins: str = "*") -> WSGIApp:
"""
Add CORS headers to every response.
Example:
app = cors_middleware(router, allowed_origins="https://myfrontend.com")
"""
def wrapped(environ: Environ, start_response: StartResponse) -> Iterable[bytes]:
def cors_sr(status: str, headers: list, *args) -> Any:
headers = list(headers)
headers.append(("Access-Control-Allow-Origin", allowed_origins))
headers.append(("Access-Control-Allow-Methods", "GET,POST,PUT,DELETE,OPTIONS"))
return start_response(status, headers, *args)
return app(environ, cors_sr)
return wrapped
# ─────────────────────────────────────────────────────────────────────────────
# 5. Dev server and test client
# ─────────────────────────────────────────────────────────────────────────────
def run_dev_server(
app: WSGIApp,
host: str = "",
port: int = 8000,
validate: bool = True,
) -> None:
"""
Start a development WSGI server (blocking).
Example:
run_dev_server(router, port=8080)
"""
wrapped = wsgiref.validate.validator(app) if validate else app
with wsgiref.simple_server.make_server(host, port, wrapped) as srv:
print(f"Serving on http://{host or 'localhost'}:{port} — Ctrl-C to stop")
srv.serve_forever()
@dataclass
class TestResponse:
status: str
headers: dict[str, str]
body: bytes
@property
def status_code(self) -> int:
return int(self.status.split()[0])
@property
def ok(self) -> bool:
return 200 <= self.status_code < 300
def text(self, encoding: str = "utf-8") -> str:
return self.body.decode(encoding)
def json(self) -> Any:
return json.loads(self.body)
def __str__(self) -> str:
return f"HTTP {self.status} ({len(self.body)} bytes)"
def call_wsgi(
app: WSGIApp,
path: str = "/",
method: str = "GET",
body: bytes | str | None = None,
headers: dict[str, str] | None = None,
query_string: str = "",
) -> TestResponse:
"""
Call a WSGI app in-process without a real HTTP server (for testing).
Example:
resp = call_wsgi(router, "/hello")
assert resp.status_code == 200
assert b"Hello" in resp.body
"""
if isinstance(body, str):
body = body.encode()
body_io = io.BytesIO(body or b"")
environ: Environ = {
"REQUEST_METHOD": method.upper(),
"PATH_INFO": path,
"QUERY_STRING": query_string,
"CONTENT_LENGTH": str(len(body or b"")),
"CONTENT_TYPE": (headers or {}).get("Content-Type", ""),
"SERVER_NAME": "localhost",
"SERVER_PORT": "80",
"SERVER_PROTOCOL": "HTTP/1.1",
"wsgi.input": body_io,
"wsgi.errors": sys.stderr,
"wsgi.url_scheme": "http",
"wsgi.multithread": False,
"wsgi.multiprocess": False,
"wsgi.run_once": False,
}
for k, v in (headers or {}).items():
key = "HTTP_" + k.upper().replace("-", "_")
environ[key] = v
response_started: list[tuple[str, list]] = []
def start_response(status: str, resp_headers: list, *args) -> None:
response_started.append((status, resp_headers))
chunks = list(app(environ, start_response))
status, resp_headers = response_started[0]
return TestResponse(
status=status,
headers={k.lower(): v for k, v in resp_headers},
body=b"".join(chunks),
)
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
print("=== wsgiref demo ===")
# ── Build a small router app ──────────────────────────────────────────────
router = Router()
@router.route("/", methods=["GET"])
def index(environ, start_response):
req = Request(environ)
return respond(start_response, f"Hello from wsgiref! uri={req.uri()}")
@router.route("/json", methods=["GET"])
def json_endpoint(environ, start_response):
params = Request(environ).query_params()
return respond_json(start_response, {"params": params, "ts": time.time()})
@router.route("/echo", methods=["POST"])
def echo(environ, start_response):
body = Request(environ).body()
return respond(start_response, body, content_type="application/octet-stream")
app = logging_middleware(cors_middleware(router))
# ── Test with call_wsgi (no socket) ───────────────────────────────────────
print("\n--- call_wsgi GET / ---")
r = call_wsgi(app, "/")
print(f" {r} body={r.text()!r}")
print("\n--- call_wsgi GET /json?foo=bar ---")
r = call_wsgi(app, "/json", query_string="foo=bar&x=1")
print(f" {r} json={r.json()}")
print("\n--- call_wsgi POST /echo ---")
r = call_wsgi(app, "/echo", method="POST", body=b"hello body")
print(f" {r} body={r.body!r}")
print("\n--- call_wsgi 404 ---")
r = call_wsgi(app, "/not-found")
print(f" {r} status_code={r.status_code}")
print("\n--- wsgiref.validate (validator middleware) ---")
try:
validated_app = wsgiref.validate.validator(router)
r = call_wsgi(validated_app, "/")
print(f" app passes wsgiref validation: {r.ok}")
except AssertionError as e:
print(f" WSGI violation: {e}")
# ── TestResponse attrs ───────────────────────────────────────────────────
print("\n--- TestResponse ---")
for path, method in [("/", "GET"), ("/json", "GET"), ("/echo", "POST")]:
r = call_wsgi(app, path, method=method, body=b"data" if method == "POST" else None)
print(f" {method} {path}: status_code={r.status_code} ok={r.ok}")
print("\n=== done (no server started — all via call_wsgi) ===")For the flask / fastapi alternative — Flask (PyPI) and FastAPI (PyPI) both implement WSGI/ASGI frameworks on top of the same PEP 3333 interface but add routing decorators, template rendering, request/response objects, and extensive extension ecosystems; FastAPI additionally generates OpenAPI schemas and uses async/await — use Flask or FastAPI for production web applications with dozens of routes and middleware needs; use wsgiref for zero-dependency microservices, test harnesses, internal tooling, and WSGI spec exploration. For the gunicorn / uvicorn alternative — gunicorn and uvicorn are production WSGI/ASGI servers that replace wsgiref.simple_server for deployment; they add process management, worker pools, graceful restart, and high-throughput I/O — the WSGI app interface remains identical; use gunicorn or uvicorn in any production deployment scenario; use wsgiref.simple_server only for development; the same WSGI callable that works with call_wsgi() above will work unchanged behind gunicorn. The Claude Skills 360 bundle includes wsgiref skill sets covering Request environ wrapper with method/path/body/json/query_params/header(), respond()/respond_json()/respond_redirect() response helpers, Router with route() decorator and not_found_handler(), logging_middleware()/cors_middleware() middleware factories, run_dev_server() with optional validator, and TestResponse/call_wsgi() in-process test client. Start with the free tier to try WSGI application patterns and wsgiref pipeline code generation.