tornado
Tornado is a Python web framework bundled with its own non-blocking HTTP server and networking layer, originally built at FriendFeed and open sourced in 2009. You subclass RequestHandler, define get and post methods, and hand a list of URL patterns to tornado.web.Application. Because it never blocks on socket I/O, one process can hold tens of thousands of open connections, which is why it became the standard answer for long polling, WebSockets, and other cases where every user keeps a socket open. Since version 5 it runs on top of asyncio, so its coroutines are plain async def functions and it shares an event loop with the rest of the async Python world.
Still a solid choice when the job is thousands of long-lived connections and you value a self-contained server with no dependencies. For a new JSON API in 2026, pick FastAPI or Starlette and stay inside the ASGI ecosystem where the tooling lives.
Use it if
- You are serving long-lived connections (WebSockets, long polling, server-sent events) to many clients at once and a thread-per-connection server would fall over
- You want an HTTP server and framework in one package with zero runtime dependencies, so the deployment artifact is just your code plus tornado
- You are extending or embedding in something already built on Tornado: Jupyter Server declares tornado>=6.2.0, so notebook extensions and kernel gateways are Tornado handlers whether you chose it or not
- You need to speak a non-HTTP protocol on the same event loop, using tornado.tcpserver or tornado.iostream alongside your web handlers
- You are building a normal JSON API: FastAPI gives you request validation, dependency injection, and OpenAPI docs that Tornado has no equivalent for, and you write far less code
- You expect the modern async Python tooling to plug in: Tornado is not an ASGI framework, so ASGI middleware, uvicorn/gunicorn worker configs, and most new observability integrations do not apply to it
- You want batteries included: there is no ORM, no migrations, no admin, no auth system, no form layer. Django ships all of that
- You need feature development, not just upkeep: the 6.5.x line through 2026 is security fixes, Python version support, and type annotations. Tornado works, but it is not growing
- Your data layer is synchronous: a blocking database driver inside a handler freezes the entire process, and routing every call through run_in_executor gives back the thread pool you were trying to avoid
Setup reality
pip install tornado pulls in literally nothing else, and wheels exist for CPython 3.9 through 3.14, so the install is a few seconds with no compiler (the optional C speedups module falls back to pure Python if a source build fails). The surprises are structural. There is no app.run() and no uvicorn command: you call app.listen(port) inside asyncio.run() and then keep the loop alive yourself with something like await asyncio.Event().wait(), which reads strangely the first time. Running one process per core means wiring bind_sockets and tornado.process.fork_processes by hand, or running N separate processes behind nginx. And the debug=True setting turns on autoreload, which interacts badly with forking, so you get one shape for development and a different one for production.
Patterns
Minimal application and serverhello-world
import asyncio
import tornado
class MainHandler(tornado.web.RequestHandler):
def get(self):
self.write("Hello, world")
async def main():
app = tornado.web.Application([(r"/", MainHandler)])
app.listen(8888)
await asyncio.Event().wait()
if __name__ == "__main__":
asyncio.run(main())listen() returns immediately; without the await asyncio.Event().wait() the coroutine finishes and asyncio.run tears the loop down before a single request arrives.
Await work inside a handlerasync-handler
from tornado.httpclient import AsyncHTTPClient
class ProxyHandler(tornado.web.RequestHandler):
async def get(self):
client = AsyncHTTPClient()
response = await client.fetch("https://api.example.com/status")
self.write(response.body)Any blocking call here (a sync DB driver, time.sleep, requests) stalls every other connection in the process; push those to IOLoop.current().run_in_executor.
Read a JSON body and return JSONjson-api
import json
class ItemsHandler(tornado.web.RequestHandler):
def post(self):
payload = json.loads(self.request.body)
self.set_status(201)
self.write({"id": 1, "name": payload["name"]})Passing a dict to write() sets Content-Type to application/json for you. Passing a list raises TypeError on purpose, because top-level JSON arrays were a historical XSS vector; wrap the list in a dict.
Path captures and query argumentsurl-arguments
class UserHandler(tornado.web.RequestHandler):
def get(self, user_id):
fields = self.get_argument("fields", "all")
tags = self.get_arguments("tag")
self.write({"id": user_id, "fields": fields, "tags": tags})
app = tornado.web.Application([
(r"/users/([0-9]+)", UserHandler),
])get_argument with no default raises a 400 MissingArgumentError; always pass a default for anything optional. Regex groups arrive as strings, so cast the id yourself.
Serve a WebSocket endpointwebsocket-handler
class EchoSocket(tornado.websocket.WebSocketHandler):
def check_origin(self, origin):
return origin == "https://app.example.com"
def open(self):
print("client connected")
async def on_message(self, message):
await self.write_message(f"echo: {message}")
def on_close(self):
print("closed", self.close_code)check_origin defaults to same-origin only, so a browser on another domain gets a 403 until you override it. Do not just return True unless the endpoint is genuinely public.
Raise HTTP errors and render them as JSONerror-handling
class ApiHandler(tornado.web.RequestHandler):
def get(self):
raise tornado.web.HTTPError(404, reason="Not found")
def write_error(self, status_code, **kwargs):
self.set_header("Content-Type", "application/json")
self.finish({"error": self._reason, "status": status_code})Without write_error, Tornado returns an HTML error page, which surprises API clients. Uncaught non-HTTPError exceptions become a 500 and are logged in full.
Gate handlers behind a current userauthenticated-routes
class BaseHandler(tornado.web.RequestHandler):
def get_current_user(self):
return self.get_signed_cookie("user")
class DashboardHandler(BaseHandler):
@tornado.web.authenticated
def get(self):
self.write(f"Hello {self.current_user.decode()}")
app = tornado.web.Application(
[(r"/dashboard", DashboardHandler)],
cookie_secret="replace-with-a-real-secret",
login_url="/login",
)The decorator redirects browsers to login_url and returns 403 for non-GET requests. get_signed_cookie returns bytes, not str, and needs cookie_secret set on the Application.
Render templates and serve static filestemplates-and-static
import os
class IndexHandler(tornado.web.RequestHandler):
def get(self):
self.render("index.html", title="Dashboard", items=[1, 2, 3])
app = tornado.web.Application(
[(r"/", IndexHandler)],
template_path=os.path.join(os.path.dirname(__file__), "templates"),
static_path=os.path.join(os.path.dirname(__file__), "static"),
)Templates autoescape by default, so use {% raw value %} for trusted HTML. static_path also enables the {{ static_url(...) }} helper, which appends a content hash for cache busting.
Flush a response incrementallystream-response
class EventsHandler(tornado.web.RequestHandler):
async def get(self):
self.set_header("Content-Type", "text/event-stream")
self.set_header("Cache-Control", "no-cache")
for i in range(10):
self.write(f"data: tick {i}\n\n")
await self.flush()
await asyncio.sleep(1)await the flush; ignoring it lets writes pile up in memory for a slow client. The connection stays open until the coroutine returns, so add a cancellation path for long streams.
Run periodic work on the same loopbackground-task
from tornado.ioloop import IOLoop, PeriodicCallback
async def sweep_stale_sessions():
...
async def main():
app = tornado.web.Application([...])
app.listen(8888)
PeriodicCallback(sweep_stale_sessions, 30_000).start()
await asyncio.Event().wait()The interval is milliseconds. PeriodicCallback skips a tick rather than overlapping if the previous run is still going, and it must be created inside a running loop.
Push blocking work to a thread pooloffload-blocking-call
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=8)
class ReportHandler(tornado.web.RequestHandler):
async def get(self):
loop = tornado.ioloop.IOLoop.current()
rows = await loop.run_in_executor(executor, slow_sync_query, "SELECT 1")
self.write({"rows": rows})Share one executor across the app instead of creating one per request. This is the escape hatch for sync database drivers, and it caps your real concurrency at max_workers.
Test handlers with a real HTTP clienttest-handlers
from tornado.testing import AsyncHTTPTestCase
class TestApi(AsyncHTTPTestCase):
def get_app(self):
return tornado.web.Application([(r"/", MainHandler)])
def test_hello(self):
response = self.fetch("/")
self.assertEqual(response.code, 200)
self.assertEqual(response.body, b"Hello, world")AsyncHTTPTestCase starts a real server on a random port for each test, so response.body is bytes. Use @tornado.testing.gen_test on the test method when you need to await inside it.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| fastapi | PyPI | Typical JSON APIs where you want validation, dependency injection, and generated OpenAPI docs |
| aiohttp | PyPI | You want async client and server in one package and prefer function handlers over handler classes |
| starlette | PyPI | A minimal ASGI toolkit that still plugs into the uvicorn and ASGI middleware ecosystem |