aiomonitor

aiomonitor is a Python 3.9+ module designed to add monitoring and CLI capabilities to asyncio applications. It works by running a task monitor in a separate thread that runs concurrently with your asyncio loop (or uvloop). This allows you to inspect the loop and provides debugging capabilities without interfering with the main application logic.

Key features include:

• Telnet server: Provides insights into your application's internal operations.
• Task management: Commands to list, cancel, and trace running asyncio tasks.
• Python REPL: Provides a Python console (via aioconsole) that executes commands directly inside your running event loop, allowing you to inspect the application state.
• GUI: A web-based interface to inspect and cancel tasks.
• Extensibility: You can add your own commands using click.

By typing console in the monitor shell, you switch to an asynchronous Python REPL. This allows you to interact with your application's state and use the await syntax to run coroutines directly within the running event loop.

monitor >>> console
>>> await asyncio.sleep(1, result=3)
3
>>> exit()

If you cannot use a context manager, you can manually control the monitor lifecycle using the Monitor class. You must call .start() to begin monitoring and, crucially, call .close() in a finally block to join the monitor thread and finalize resources.

m = aiomonitor.Monitor(loop)
m.start()
try:
    loop.run_forever()
finally:
    m.close()

To monitor an aiohttp application, use aiomonitor.start_monitor() within an asyncio context. You should initialize the monitor just before running your application. Passing hook_task_factory=True allows you to see recursive task creation and termination history in the Web-based Inspector.

import asyncio
import aiomonitor
from aiohttp import web

async def simple(request):
    await asyncio.sleep(100)
    return web.Response(text="Simple answer")

async def main():
   app = web.Application()
   app.router.add_get("/simple", simple)

   loop = asyncio.get_running_loop()
   # Initialize monitor just before running the app
   with aiomonitor.start_monitor(loop, hook_task_factory=True):
       await web._run_app(app, port=8090, host="localhost")

if __name__ == "__main__":
    asyncio.run(main())

The simplest way to start an aiomonitor session is by using the aiomonitor.start_monitor(loop) factory function as a context manager. This automatically handles the lifecycle of the monitor, ensuring it starts before your loop runs and cleans up resources when the context is exited. You can connect to the resulting monitor via telnet (e.g., telnet localhost 20101).

import asyncio
import aiomonitor

async def main():
    loop = asyncio.get_event_loop()
    with aiomonitor.start_monitor(loop):
        print("Now you can connect with: telnet localhost 20101")
        loop.run_forever()

asyncio.run(main())

Once the application is running, you can connect to the monitor from a separate terminal.

Using Telnet/Netcat: Note: Since version 0.5.0, you must use a telnet client that implements the actual telnet protocol to support advanced features like auto-completion.

telnet localhost 20101

Using the built-in CLI: If you do not have a telnet client installed, aiomonitor provides its own CLI:

python -m aiomonitor.cli

You can view a web-based UI to inspect currently running and terminated tasks, including their recursive stack traces, by navigating to http://localhost:10202 (or the port configured). You can also cancel specific tasks from this UI.

To enable recursive task creation and termination history in the inspector, ensure you pass hook_task_factory=True to aiomonitor.start_monitor().

Once the monitor is running, you can connect to it from a separate terminal using either telnet or the built-in aiomonitor.cli module.

# Using telnet (default port is usually 20101)
$ telnet localhost 20101

# Using the built-in python client
$ python -m aiomonitor.cli

Install aiomonitor using pip. It requires Python 3.8+ (3.10.7+ recommended) and depends on aioconsole, click, and prompt_toolkit. uvloop is an optional dependency.

$ pip install aiomonitor