Python and asyncio minimum

Learning objectives

After this lesson, you should be able to read the following constructs in nanobot’s core flow:

• dataclass;
• abstract base classes;
• async def and await;
• asyncio.Task, Queue, and Lock;
• timeouts and cancellation;
• async with;
• callbacks and type annotations.

This is not a complete asyncio tutorial. It explains only the mental models required by the later source lessons.

Prerequisites

You only need to read ordinary Python functions, classes, lists, dictionaries, exceptions, and file operations.

1. A coroutine is not an already-computed result

Calling a normal function executes it immediately:

result = normal_function()

Calling an async def function produces a coroutine. It progresses only when you await it, wrap it in a Task, or hand it to the event loop:

result = await async_function()

Provider requests, tool execution, message queues, and Channel sends in nanobot may wait for a network, subprocess, or another task, so many interfaces are asynchronous.

Reading technique

When you find an asynchronous function, do not stop at its definition. Search for callers that await it or wrap it in a Task. The real concurrency boundary usually sits at the caller.

2. A Task is independently progressing asynchronous work

task = asyncio.create_task(handle_message(message))

A Task lets the event loop continue other work while one operation waits for a network or tool.

AgentLoop.run() does not block the entire message-consumption loop after receiving one message. It creates a Task for a message that can be handled independently. At the same time, a session lock prevents competing turns within one session.

A useful first model is:

Across sessions: may run concurrently
Within one session: must be serialized
A new message during one turn: may enter a pending queue

Source to revisit later:

• nanobot/agent/loop.py
  • AgentLoop.run
  • AgentLoop._dispatch

3. Queue connects asynchronous producers and consumers

asyncio.Queue is useful when asynchronous components work at different rates.

await queue.put(message)
message = await queue.get()

Inside nanobot, MessageBus maintains two queues:

• inbound: messages sent from a Channel or another entry point to the Agent;
• outbound: messages sent from the Agent back to a Channel.

Queue solves a handoff problem. It is not the same thing as:

• Session;
• event history;
• RuntimeEventBus;
• persistent storage.

Related source:

• nanobot/bus/queue.py
• nanobot/bus/events.py

4. Lock protects a critical section

async with lock:
    await update_session()

If two turns in one session read, modify, and save history at the same time, their ordering can break or one update can overwrite another. nanobot keeps an asyncio.Lock for each effective session key and handles the message only after _dispatch() acquires it.

Lock does not mean the whole program is single-threaded

It constrains only code using that lock. Different sessions can still progress concurrently, and tools may also execute concurrently.

5. Semaphore limits total concurrency

Lock answers “who may enter this critical section together?” Semaphore answers “how many tasks may enter at once?”

nanobot can use NANOBOT_MAX_CONCURRENT_REQUESTS as a request-concurrency gate. Verify the current default and initialization in AgentLoop rather than relying on memory.

6. Timeout and cancellation are not ordinary errors

A common pattern is:

result = await asyncio.wait_for(operation(), timeout=30)

A timeout commonly triggers cancellation. asyncio.CancelledError requires careful handling:

• real cancellation must continue propagating during normal shutdown;
• code must not swallow every cancellation and continue running;
• cleanup of network connections, MCP transports, and background work must respect the Task that owns each resource.

Useful places to observe in nanobot:

• timeout and cancellation around queue consumption in AgentLoop.run();
• the LLM timeout in AgentRunner._request_model();
• Provider stream idle timeout;
• /stop and task cancellation;
• MCP connection shutdown.

7. async with manages asynchronous resources

async with lock:
    ...

It is the asynchronous entry-and-exit protocol. Exit logic still runs if the body raises an exception. nanobot uses it for:

• session locks;
• the concurrency gate;
• HTTP requests;
• selected transports or streaming responses.

8. dataclass makes runtime data explicit

@dataclass
class Message:
    channel: str
    content: str

nanobot uses dataclasses for runtime data such as:

• InboundMessage;
• OutboundMessage;
• AgentRunSpec;
• AgentRunResult;
• runtime events.

When reading one, ask:

1. Who creates it?
2. Who reads it?
3. Is it written to disk?
4. Are its fields part of a stable public contract?

Do not assume that an object is persistent or public merely because it is a dataclass.

9. An abstract base class defines a minimum contract

Tool, LLMProvider, and BaseChannel use abstract interfaces to constrain implementations.

An abstract base class says what an implementation must provide; concrete classes still determine actual behavior. For example:

• Tool.execute() makes tool execution asynchronous;
• Tool.concurrency_safe determines whether a tool may join a concurrent tool batch;
• BaseChannel.start() must keep a Channel running;
• LLMProvider.chat() returns a normalized LLMResponse.

10. Callbacks hand events back to the caller

Streaming output and progress updates commonly use callbacks:

async def on_stream(delta: str) -> None:
    ...

The core does not need to know whether a terminal, HTTP SSE response, or WebSocket ultimately consumes the delta. The caller supplies a callback, allowing the same AgentRunner to serve multiple surfaces.

This is an important clue to nanobot’s boundaries:

Core modules produce general events; an entry point or Channel decides how to present them.

Map the concepts to nanobot

Python concept	nanobot location
dataclass	nanobot/bus/events.py, nanobot/agent/runner.py
Queue	nanobot/bus/queue.py
Task	AgentLoop.run()
Lock	AgentLoop._dispatch(), API session locks
Semaphore	the AgentLoop concurrency gate
callback	stream, progress, and retry-wait callbacks
cancellation	AgentLoop.run(), task stop, and shutdown
abstract base class	Tool, Provider, and Channel

Self-check

Explain in your own words:

1. Why does MessageBus use a Queue while Session does not?
2. Why does nanobot allow concurrency across sessions while locking each individual session?
3. After asyncio.create_task(), why must code still retain the Task or handle its exception?
4. Why should CancelledError not always be treated like an ordinary Exception?

If you can answer these questions, you know enough to continue into the core flow.

Next step

Read LLM messages, streaming, and tool calling.