async #
python's async/await syntax is straightforward. the interesting part is how you structure code around it.
async with #
the core insight from async python codebases: async with is how you manage resources. not try/finally, not callbacks - the context manager protocol.
when you open a connection, start a session, or acquire any resource that needs cleanup, you wrap it in an async context manager:
@asynccontextmanager
async def get_atproto_client(
require_auth: bool = False,
operation: str = "this operation",
target_repo: str | None = None,
) -> AsyncIterator[AsyncClient]:
"""get an atproto client using credentials from context or environment."""
client = AsyncClient(pds_url)
if require_auth and handle and password:
await client.login(handle, password)
try:
yield client
finally:
pass # AsyncClient doesn't need explicit cleanup
the caller writes async with get_atproto_client() as client: and cleanup happens automatically. this pattern appears constantly - database connections, HTTP sessions, file handles, locks.
from pdsx/mcp/client.py (pdsx is an atproto record CLI + MCP server)
the alternative - manual try/finally blocks scattered through the code, or worse, forgetting cleanup entirely - is why this pattern dominates. you encode the lifecycle once in the context manager, and every use site gets it right by default.
ContextVar #
python added contextvars to solve a specific problem: how do you have request-scoped state in async code without passing it through every function?
in sync code, you might use thread-locals. but async tasks can interleave on the same thread, so thread-locals don't work. ContextVar gives each task its own copy:
from contextvars import ContextVar
_current_docket: ContextVar[Docket | None] = ContextVar("docket", default=None)
_current_worker: ContextVar[Worker | None] = ContextVar("worker", default=None)
_current_server: ContextVar[weakref.ref[FastMCP] | None] = ContextVar("server", default=None)
set it at the start of handling a request, and any code called from that task can access it. this is how frameworks like fastapi and fastmcp pass request context without threading it through every function signature.
the pattern: set at the boundary (request handler, task entry), read anywhere inside. reset when you're done.
from fastmcp/server/dependencies.py
concurrency control #
asyncio.gather() runs tasks concurrently, but sometimes you need to limit how many run at once - rate limits, connection pools, memory constraints.
asyncio.Semaphore is the primitive for this. acquire before work, release after. the async with syntax makes it clean:
semaphore = asyncio.Semaphore(concurrency)
async def delete_one(uri: str) -> None:
"""delete a single record with concurrency control."""
async with semaphore:
try:
await delete_record(client, uri)
successful.append(uri)
except Exception as e:
failed.append((uri, e))
if fail_fast:
raise
await asyncio.gather(*[delete_one(uri) for uri in uris])
at most concurrency delete operations run at once. the rest wait.
database clients #
connection pooling and batch-write patterns with asyncpg live in storage/postgres.