PyPIWeb Backendupdated 06 Aug 2026

trio

Item: trio
Rating: 4
Author: Keyoor

Trio is an async/await runtime for Python built around structured concurrency: every background task must be spawned inside a nursery, and the nursery block does not exit until all of its children have finished. That single rule removes the whole category of bugs where a task is created, forgotten, and later dies unobserved. On top of it Trio ships cancel scopes (composable timeouts that work even across library boundaries), memory channels, thread bridges, TCP and TLS streams, subprocess helpers, and a testing module with a fake clock. It is a complete alternative to asyncio, not a layer on top of it, so the two event loops do not mix without a bridge.

trio · v0.33.0 · 12.2M/wk · ★ 7,307 · pushed 03 Aug 2026 · python-trio/trio · docs · reviewed by Keyoor

Verdict

The best-designed concurrency model in Python and the one that makes cancellation and task lifetime obvious rather than accidental, but you pay for it in drivers you cannot use and deployment paths nobody has documented. Choose Trio for self-contained applications and protocol work; choose anyio if the code is a library other people will import.

API stability4/5Core primitives such as run, open_nursery, and the cancel scopes have not changed in years and deprecations get long warning periods, but the project is still pre-1.0 and did land real migrations recently: MultiError removed in favour of ExceptionGroup, and strict exception groups becoming the default.

Docs5/5trio.readthedocs.io is among the best documentation in Python: a tutorial that teaches async from zero, a reference that explains why each primitive exists, an explicit section on cancellation and checkpoints, and a written history page for every release.

Maintenance4/5Pushed August 2026 with 0.33.0 released February 2026, a real team of maintainers rather than one person, and a public forum. The counterweight is about 277 open issues and a slowing release pace compared with the project's early years.

Ecosystem3/5Roughly 12.2M downloads a week, but the bulk of that is transitive through anyio-based stacks rather than direct use. The trio-* library shelf (trio-websocket, asyncclick, hypercorn, httpx via anyio) covers common cases; databases and cloud SDKs mostly do not.

Use it if

You are writing a network client or server from scratch and correctness under cancellation matters more than reusing existing asyncio libraries: nurseries make it very hard to leak a task, and cancel scopes make timeouts compose instead of fighting each other
You want to test time-dependent code fast: trio.testing.MockClock with autojump makes a program that sleeps for an hour finish instantly, with no sleep patching
Your concurrency is genuinely tricky (retries, races, happy eyeballs, supervision trees) and you would rather express it in nurseries and scopes than in asyncio.gather plus manual task bookkeeping
You are already writing against anyio and want the original design with better error messages and a smaller surface to reason about

Skip it if

You need the mainstream async library ecosystem. asyncpg, aiohttp, aiomysql, aiobotocore, motor, and the async halves of SQLAlchemy and redis-py are asyncio-only; on Trio you either find a Trio-native port, run them through trio-asyncio, or write the driver yourself
You are deploying a web app the usual way. uvicorn is asyncio, and while Hypercorn and Quart can run on Trio, nearly all deployment guides, ASGI middleware, and hosting docs assume asyncio underneath
You are writing a library rather than an application. anyio lets the same code serve both asyncio and Trio users, which is why httpx and Starlette chose it; committing a library to Trio alone cuts off most of your potential users
Your team is not ready for exception groups. Since strict mode became the default, a nursery wraps even a single failure in an ExceptionGroup, so plain except ValueError stops matching and you need except* or the exceptiongroup helpers on Python 3.10
You expect a settled 1.0. After nine years it is still on 0.x and classified Development Status 4 (Beta), with roughly 277 open issues; breaking changes are rare and well announced, but the project itself tells you to subscribe to issue #1 if you depend on API stability

Setup reality

pip install trio is painless: pure Python dependencies (attrs, sortedcontainers, idna, outcome, sniffio) plus cffi on Windows where a wheel already exists, so no compiler is needed, and Python 3.10 is the floor. The cost lands later. Trio is a different runtime, so anything that calls asyncio.get_event_loop underneath will not work, and finding that out often means reading a dependency's source. Cancellation only takes effect at checkpoints, so a function with no await in its loop is uninterruptible and a function that never awaits can starve everything else; the library ships trio.lowlevel.checkpoint() precisely for that. Add pytest-trio for tests, because plain pytest cannot run an async def test on its own, and expect to spend the first week unlearning create_task and asyncio.gather.

Patterns

Run concurrent tasks in a nurseryrun-tasks-in-a-nursery

import trio

async def worker(name: str) -> None:
    await trio.sleep(1)
    print("done", name)

async def main() -> None:
    async with trio.open_nursery() as nursery:
        nursery.start_soon(worker, "a")
        nursery.start_soon(worker, "b")
    print("both finished")

trio.run(main)

The async with block will not exit until every child returns, so tasks cannot outlive their nursery. start_soon takes positional arguments only; use functools.partial for keywords.

Apply a timeout that composestimeouts-with-cancel-scopes

import trio

async def main() -> None:
    with trio.move_on_after(5) as scope:
        await slow_call()
    if scope.cancelled_caught:
        print("gave up after 5s")

    with trio.fail_after(5):        # raises trio.TooSlowError
        await slow_call()

    # extend a deadline from inside
    with trio.move_on_after(1) as scope:
        scope.deadline = trio.current_time() + 30
        await slow_call()

These are ordinary with statements, not async with. Cancellation lands only at a checkpoint, so a CPU-bound loop with no await runs to completion regardless of the deadline.

Catch failures coming out of a nurseryhandle-exception-groups

import trio

async def main() -> None:
    try:
        async with trio.open_nursery() as nursery:
            nursery.start_soon(might_fail)
            nursery.start_soon(might_also_fail)
    except* ValueError as group:      # Python 3.11+
        for exc in group.exceptions:
            print("task failed:", exc)
    except* OSError:
        print("network problem")

Since strict exception groups became the default, even a single failing task arrives wrapped, so a plain except ValueError no longer matches. On Python 3.10 use exceptiongroup.catch instead of except* syntax.

Pass work between tasks with backpressurememory-channels

import trio

async def producer(send_channel) -> None:
    async with send_channel:
        for i in range(100):
            await send_channel.send(i)

async def consumer(receive_channel) -> None:
    async with receive_channel:
        async for item in receive_channel:
            await handle(item)

async def main() -> None:
    send, receive = trio.open_memory_channel(max_buffer_size=10)
    async with trio.open_nursery() as nursery:
        nursery.start_soon(producer, send)
        nursery.start_soon(consumer, receive)

trio.run(main)

Closing the send side is what ends the consumer's async for loop. With several consumers, give each one its own clone via receive.clone() and close the original, otherwise the loop never terminates.

Call blocking code without stalling the looprun-blocking-code-in-a-thread

import functools
import trio

async def main() -> None:
    rows = await trio.to_thread.run_sync(query_database, "select 1")

    # keyword arguments need partial
    body = await trio.to_thread.run_sync(
        functools.partial(requests.get, url, timeout=10)
    )

    # let the thread keep running if we are cancelled
    await trio.to_thread.run_sync(flush_to_disk, abandon_on_cancel=True)

By default a cancelled to_thread.run_sync waits for the thread to finish, because Python cannot interrupt arbitrary C code. abandon_on_cancel=True returns immediately but leaves the thread running, so only use it for work that is safe to orphan.

Call back into Trio from a worker threadcall-trio-from-a-thread

import trio

def blocking_worker(paths):
    for path in paths:
        data = process(path)
        trio.from_thread.run(report_progress, path)   # await an async fn
        trio.from_thread.run_sync(counter.increment)  # sync fn on the loop

async def main() -> None:
    await trio.to_thread.run_sync(blocking_worker, paths)

from_thread only works inside a thread Trio itself started. For a thread you did not create, open a trio.from_thread.start_blocking_portal and hand the portal to it.

Serve and connect over TCPtcp-server-and-client

import trio

async def echo(stream: trio.SocketStream) -> None:
    async with stream:
        async for chunk in stream:
            await stream.send_all(chunk)

async def server() -> None:
    await trio.serve_tcp(echo, 9000)

async def client() -> None:
    stream = await trio.open_tcp_stream("127.0.0.1", 9000)
    async with stream:
        await stream.send_all(b"hello\n")
        print(await stream.receive_some(1024))

open_tcp_stream with a hostname runs happy eyeballs across IPv4 and IPv6 for you. serve_tcp runs forever and opens an internal nursery, so wrap it in a cancel scope if you need to shut it down.

Start a task and wait until it is readywait-for-task-startup

import trio

async def serve(port: int, *, task_status=trio.TASK_STATUS_IGNORED) -> None:
    listeners = await trio.open_tcp_listeners(port)
    task_status.started(listeners[0].socket.getsockname())
    await trio.serve_listeners(handler, listeners)

async def main() -> None:
    async with trio.open_nursery() as nursery:
        address = await nursery.start(serve, 0)
        print("bound to", address)  # guaranteed listening now

nursery.start blocks until the child calls task_status.started(); if the child raises before that, the error propagates to the caller instead of leaving a half-started service behind.

Let cleanup finish even while cancelledshield-cleanup-from-cancellation

import trio

async def handle(conn) -> None:
    try:
        await serve_requests(conn)
    finally:
        with trio.CancelScope(shield=True) as scope:
            scope.deadline = trio.current_time() + 2
            await conn.send_goodbye()
            await conn.aclose()

Without the shield, any await inside finally is cancelled instantly and the connection leaks. Always pair a shield with its own deadline or a hung peer turns shutdown into a hang.

Run a subprocess with a timeoutrun-a-subprocess

import trio

async def main() -> None:
    with trio.move_on_after(30):
        result = await trio.run_process(
            ["ffmpeg", "-i", "in.mov", "out.mp4"],
            capture_stdout=True,
            capture_stderr=True,
        )
        print(result.returncode, result.stdout[:200])

run_process raises CalledProcessError on a non-zero exit unless you pass check=False, and on cancellation it kills the child for you. For streaming stdin and stdout while the process runs, use trio.lowlevel.open_process inside a nursery.

Cap how many tasks run at oncelimit-concurrency

import trio

async def main(urls) -> None:
    limiter = trio.CapacityLimiter(8)

    async def fetch(url: str) -> None:
        async with limiter:
            await download(url)

    async with trio.open_nursery() as nursery:
        for url in urls:
            nursery.start_soon(fetch, url)

Spawning ten thousand tasks is cheap, but ten thousand simultaneous sockets is not; the limiter throttles inside the tasks rather than throttling how many you start. trio.Semaphore exists too, but CapacityLimiter reports statistics and is the recommended choice.

Test timeouts without waiting for real timetest-with-a-fake-clock

import pytest
import trio
from trio.testing import MockClock, wait_all_tasks_blocked

@pytest.mark.trio
async def test_retry_backoff(autojump_clock):
    start = trio.current_time()
    await retry_with_backoff(always_fails, attempts=5)
    assert trio.current_time() - start == pytest.approx(31.0)

def test_standalone():
    trio.run(main, clock=MockClock(autojump_threshold=0))

The autojump_clock fixture comes from pytest-trio, which you need installed for @pytest.mark.trio to work at all. The clock only skips ahead when every task is blocked, so a busy loop makes the test hang instead of jumping.

Alternatives

Package	Registry	Pick it when
anyio	PyPI	You are writing a library and need the same code to run for both asyncio and Trio users.
trio-asyncio	PyPI	You want Trio's model but must call one or two asyncio-only drivers from inside it.
uvloop	PyPI	You are staying on asyncio for ecosystem reasons and just want a faster event loop.
curio	PyPI	You want the smaller research-flavoured runtime that inspired Trio, accepting a far quieter project.