### Start Monitor with Context Manager

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/api.md

Use the `aiomonitor.start_monitor` context manager for a simple way to start and automatically stop the monitor. This is the recommended approach for most use cases.

```python
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())
```

--------------------------------

### Install Development Dependencies

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/contributing.md

Install the necessary dependencies for development and the aiomonitor package in editable mode.

```bash
pip install -r requirements-dev.txt
```

--------------------------------

### Basic aiohttp Server with aiomonitor

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/tutorial.md

Sets up a simple aiohttp web server and integrates aiomonitor to monitor tasks. Ensure aiomonitor is installed.

```python
import asyncio

import aiomonitor
from aiohttp import web

# A simple handler that returns response after 100s
async def simple(request):
    print("Start sleeping")
    await asyncio.sleep(100)
    return web.Response(text="Simple answer")

async def main():
   # create application and register route create route
   app = web.Application()
   app.router.add_get("/simple", simple)

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

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

```

--------------------------------

### Start aiomonitor using start_monitor context manager

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Use `start_monitor` as a context manager to automatically start and stop the aiomonitor. Configure ports, task tracking, and variables exposed to the REPL.

```python
import asyncio
import aiomonitor

async def worker():
    while True:
        await asyncio.sleep(1)

async def main():
    loop = asyncio.get_running_loop()

    # Start the monitor: telnet UI on :20101, web UI on :20102, REPL on :20103
    with aiomonitor.start_monitor(
        loop,
        host="127.0.0.1",
        port=20101,            # terminal UI port (also: termui_port)
        webui_port=20102,      # web inspector port
        console_port=20103,    # async Python REPL port
        hook_task_factory=True,  # enable task creation/cancellation chain tracking
        max_termination_history=500,  # how many terminated tasks to remember
        locals={"app_version": "1.0"},  # variables exposed in REPL console
    ) as monitor:
        print(f"Monitor running at telnet://{monitor.host}:{monitor.port}")
        asyncio.create_task(worker())
        await asyncio.sleep(3600)  # run forever

try:
    asyncio.run(main())
except KeyboardInterrupt:
    pass

# Connect from another terminal:
#   telnet localhost 20101
#   python -m aiomonitor.cli -H 127.0.0.1 -p 20101
```

--------------------------------

### Start Basic Monitor

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/examples.md

Starts a basic aiomonitor instance attached to the asyncio event loop. Use this for simple debugging needs.

```python
import asyncio

import aiomonitor


async def main() -> None:
    loop = asyncio.get_running_loop()
    with aiomonitor.start_monitor(loop=loop) as monitor:
        print("Now you can connect with:")
        print(f"  python -m aiomonitor.cli -H {monitor.host} -p {monitor.port}")
        print("or")
        print(f"  telnet {monitor.host} {monitor.port}  # Linux only")
        loop.run_forever()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass
```

--------------------------------

### Start aiomonitor with Context Manager

Source: https://github.com/aio-libs/aiomonitor/blob/main/README.rst

Use the `start_monitor` context manager to easily integrate aiomonitor into your asyncio application. This ensures the monitor is properly started and stopped.

```python
import asyncio

import aiomonitor

async def main():
    loop = asyncio.get_running_loop()
    run_forever = loop.create_future()
    with aiomonitor.start_monitor(loop):
        await run_forever

try:
    asyncio.run(main())
except KeyboardInterrupt:
    pass
```

--------------------------------

### Clone aiomonitor Repository

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/contributing.md

Clone the aiomonitor repository to start contributing. Ensure you have Git installed.

```bash
git clone git@github.com:aio-libs/aiomonitor.git
```

--------------------------------

### Initialize and manage Monitor class directly

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Instantiate the `Monitor` class directly for manual control over starting and closing. This allows for explicit management of the monitor's lifecycle.

```python
import asyncio
import aiomonitor
from aiomonitor.monitor import Monitor

async def main():
    loop = asyncio.get_running_loop()

    monitor = Monitor(
        loop,
        host="127.0.0.1",
        termui_port=20101,
        webui_port=20102,
        console_port=20103,
        console_enabled=True,
        hook_task_factory=True,
        max_termination_history=1000,
        locals={"loop": loop},
    )
    monitor.start()
    print(f"Monitor started: {monitor!r}")  # <Monitor: 127.0.0.1:20101>
    print(f"Closed? {monitor.closed}")      # False

    try:
        await asyncio.sleep(3600)
    finally:
        monitor.close()   # joins background thread, restores original task factory
        print(f"Closed? {monitor.closed}")  # True

asyncio.run(main())
```

--------------------------------

### Running the aiohttp Server

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/tutorial.md

Command to run the Python script that starts the aiohttp server with aiomonitor.

```bash
$ python simple_srv.py
======== Running on http://localhost:8090 ========
(Press CTRL+C to quit)

```

--------------------------------

### aiomonitor.monitor.start_monitor()

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/monitor.md

Factory function to create and start an instance of the Monitor. It allows configuration of host, ports for terminal UI, web UI, and console, as well as enabling the console and providing local variables.

```APIDOC
## aiomonitor.monitor.start_monitor()

### Description
Factory function, creates instance of [`Monitor`](#aiomonitor.monitor.Monitor) and starts monitoring thread.

### Parameters
* **monitor** (*Type* *[*[*Monitor*](#aiomonitor.monitor.Monitor) *]*) – Monitor class to use
* **host** (*str*) – hostname to serve monitor telnet server
* **port** (*int*) – monitor port (terminal UI), by default 20101
* **webui_port** (*int*) – monitor port (web UI), by default 20102
* **console_port** (*int*) – python REPL port, by default 20103
* **console_enabled** (*bool*) – flag indicates if python REPL is requred to start with instance of monitor.
* **locals** (*dict*) – dictionary with variables exposed in python console environment
```

--------------------------------

### aiohttp with Custom Monitor Command

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/examples.md

Extends aiomonitor with a custom command ('hello') for an aiohttp application. This example demonstrates how to define and use custom commands for interacting with the monitored application.

```python
import asyncio
from typing import Optional

import click
import requests
import uvloop
from aiohttp import web

import aiomonitor
from aiomonitor.termui.commands import (
    auto_command_done,
    custom_help_option,
    monitor_cli,
)


async def simple(request: web.Request) -> web.Response:
    await asyncio.sleep(10)
    return web.Response(text="Simple answer")


async def hello(request: web.Request) -> web.StreamResponse:
    resp = web.StreamResponse()
    name = request.match_info.get("name", "Anonymous")
    answer = ("Hello, " + name).encode("utf8")
    resp.content_length = len(answer)
    resp.content_type = "text/plain"
    await resp.prepare(request)
    await asyncio.sleep(10)
    await resp.write(answer)
    await resp.write_eof()
    return resp


@monitor_cli.command(name="hello")
@click.argument("name", optional=True)
@custom_help_option
@auto_command_done
def do_hello(ctx: click.Context, name: Optional[str] = None) -> None:
    """Using the /hello GET interface

    There is one optional argument, "name".  This name argument must be
    provided with proper URL escape codes, like %20 for spaces.
    """
    name = "" if name is None else "/" + name
    r = requests.get("http://localhost:8090/hello" + name)
    click.echo(r.text + "\n")


async def main() -> None:
    loop = asyncio.get_running_loop()
    app = web.Application()
    app.router.add_get("/simple", simple)
    app.router.add_get("/hello/{name}", hello)
    app.router.add_get("/hello", hello)
    host, port = "localhost", 8090
    with aiomonitor.start_monitor(loop, locals=locals()):
        web.run_app(app, port=port, host=host)


if __name__ == "__main__":
    uvloop.install()
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass
```

--------------------------------

### Start Monitor Manually

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/api.md

Alternatively, manually start and close the monitor using the `Monitor` class. Ensure `close()` is called in a `finally` block to release resources.

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

--------------------------------

### Asyncio Sleep Example

Source: https://github.com/aio-libs/aiomonitor/blob/main/README.rst

Demonstrates the usage of asyncio.sleep with a result. This is typically run within an asyncio event loop.

```python
>>> await asyncio.sleep(1, result=3)
3
>>>
```

--------------------------------

### start_monitor

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Starts the aiomonitor in a background daemon thread using a context manager for automatic cleanup. It provides telnet, web, and REPL interfaces for monitoring.

```APIDOC
## start_monitor(loop, ...)

### Description
Factory function that creates and starts a `Monitor` instance in a background daemon thread. Supports context manager protocol for automatic cleanup. Returns the `Monitor` instance.

### Parameters
- **loop** (asyncio.AbstractEventLoop) - The asyncio event loop to monitor.
- **host** (str) - The host address for the monitor interfaces.
- **port** (int) - The port for the terminal UI (telnet).
- **webui_port** (int) - The port for the web inspector UI.
- **console_port** (int) - The port for the async Python REPL console.
- **hook_task_factory** (bool) - If True, enables tracking of task creation and cancellation chains.
- **max_termination_history** (int) - The number of terminated tasks to remember.
- **locals** (dict) - A dictionary of variables to expose in the REPL console.

### Request Example
```python
import asyncio
import aiomonitor

async def worker():
    while True:
        await asyncio.sleep(1)

async def main():
    loop = asyncio.get_running_loop()

    with aiomonitor.start_monitor(
        loop,
        host="127.0.0.1",
        port=20101,            # terminal UI port (also: termui_port)
        webui_port=20102,      # web inspector port
        console_port=20103,    # async Python REPL port
        hook_task_factory=True,  # enable task creation/cancellation chain tracking
        max_termination_history=500,  # how many terminated tasks to remember
        locals={"app_version": "1.0"},  # variables exposed in REPL console
    ) as monitor:
        print(f"Monitor running at telnet://{monitor.host}:{monitor.port}")
        asyncio.create_task(worker())
        await asyncio.sleep(3600)  # run forever

try:
    asyncio.run(main())
except KeyboardInterrupt:
    pass
```

### Response
- **monitor** (Monitor) - The Monitor instance.
```

--------------------------------

### Interact with aiohttp in Python REPL

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/tutorial.md

Example of using aiohttp within the aiomonitor's Python REPL to make an HTTP request and inspect the response.

```python
>>> import aiohttp
>>> session = aiohttp.ClientSession()
>>> resp = await session.get('http://python.org')
>>> resp.status
200
>>> data = await resp.read()
>>> len(data)
47373
>>>

```

--------------------------------

### Integrate aiomonitor with aiohttp Web Application

Source: https://github.com/aio-libs/aiomonitor/blob/main/README.rst

Integrate aiomonitor into an aiohttp application by starting the monitor before running the application. You can pass local variables to the monitor's console environment.

```python
import asyncio

import aiomonitor
from aiohttp import web

# Simple handler that returns response after 100s
async def simple(request):
    print('Start sleeping')
    await asyncio.sleep(100)
    return web.Response(text="Simple answer")

loop = asyncio.get_event_loop()
# create application and register route
app = web.Application()
app.router.add_get('/simple', simple)

# it is possible to pass a dictionary with local variables
# to the python console environment
host, port = "localhost", 8090
locals_ = {"port": port, "host": host}
# init monitor just before run_app
with aiomonitor.start_monitor(loop=loop, locals=locals_):
    # run application with built-in aiohttp run_app function
    web.run_app(app, port=port, host=host, loop=loop)
```

--------------------------------

### Monitor Class

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

The core monitoring class. Can be used directly with explicit `start()` / `close()` calls instead of the context manager. Exposes `host`, `port`, and `console_locals` as public attributes.

```APIDOC
## Monitor(loop, ...)

### Description
The core monitoring class. Can be used directly with explicit `start()` / `close()` calls instead of the context manager. Exposes `host`, `port`, and `console_locals` as public attributes.

### Parameters
- **loop** (asyncio.AbstractEventLoop) - The asyncio event loop to monitor.
- **host** (str) - The host address for the monitor interfaces.
- **termui_port** (int) - The port for the terminal UI (telnet).
- **webui_port** (int) - The port for the web inspector UI.
- **console_port** (int) - The port for the async Python REPL console.
- **console_enabled** (bool) - Whether the console is enabled.
- **hook_task_factory** (bool) - If True, enables tracking of task creation and cancellation chains.
- **max_termination_history** (int) - The number of terminated tasks to remember.
- **locals** (dict) - A dictionary of variables to expose in the REPL console.

### Methods
- **start()**: Starts the monitor in a background thread.
- **close()**: Stops the monitor, joins the background thread, and restores the original task factory.

### Attributes
- **host** (str) - The host address.
- **port** (int) - The terminal UI port.
- **console_locals** (dict) - The dictionary of variables exposed in the REPL console.
- **closed** (bool) - Indicates if the monitor is closed.

### Request Example
```python
import asyncio
import aiomonitor
from aiomonitor.monitor import Monitor

async def main():
    loop = asyncio.get_running_loop()

    monitor = Monitor(
        loop,
        host="127.0.0.1",
        termui_port=20101,
        webui_port=20102,
        console_port=20103,
        console_enabled=True,
        hook_task_factory=True,
        max_termination_history=1000,
        locals={"loop": loop},
    )
    monitor.start()
    print(f"Monitor started: {monitor!r}")  # <Monitor: 127.0.0.1:20101>
    print(f"Closed? {monitor.closed}")      # False

    try:
        await asyncio.sleep(3600)
    finally:
        monitor.close()   # joins background thread, restores original task factory
        print(f"Closed? {monitor.closed}")  # True

asyncio.run(main())
```
```

--------------------------------

### Add Synchronous Custom Command to Monitor CLI

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/tutorial.md

Define a synchronous command for the monitor CLI using `@monitor_cli.command`. Ensure `auto_command_done` is used to notify the main loop of completion. This example demonstrates making an HTTP request using the `requests` library.

```python
import aiohttp
import click
import requests
from aiomonitor.termui.commands import (
    auto_async_command_done,
    auto_command_done,
    custom_help_option,
    monitor_cli,
)

@monitor_cli.command(name="hello")
@click.argument("name", optional=True)
@custom_help_option
@auto_command_done  # sync version
def do_hello(ctx: click.Context, name: Optional[str] = None) -> None:
    """An example command to say hello to another HTTP server."""
    name = "unknown" if name is None else name
    r = requests.get("http://example.com/hello/" + name)
    click.echo(r.text + "\n")
```

--------------------------------

### Add Asynchronous Custom Command to Monitor CLI

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/tutorial.md

Define an asynchronous command for the monitor CLI using `@monitor_cli.command`. Use `@auto_async_command_done` for asynchronous completion notification. This example uses `aiohttp` for making an asynchronous HTTP request.

```python
import aiohttp
import click
import requests
from aiomonitor.termui.commands import (
    auto_async_command_done,
    auto_command_done,
    custom_help_option,
    monitor_cli,
)

@monitor_cli.command(name="hello-async")
@click.argument("name", optional=True)
@custom_help_option
@auto_async_command_done  # async version
async def do_async_hello(ctx: click.Context, name: Optional[str] = None) -> None:
    """An example command to asynchronously say hello to another HTTP server."""
    name = "unknown" if name is None else name
    async with aiohttp.ClientSession() as sess:
        async with sess.get("http://example.com/hello/" + name) as resp:
            click.echo(await resp.text())
```

--------------------------------

### Set Up Development Virtual Environment

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/contributing.md

Create and activate a Python virtual environment for development using pyenv. Requires Python 3.8 or later.

```bash
cd aiomonitor
pyenv virtualenv 3.11 aiomonitor-dev
pyenv local aiomonitor-dev
```

--------------------------------

### aiomonitor.webui.app.init_webui

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/webui.md

Initializes the web UI application with the provided monitor instance.

```APIDOC
## async init_webui(monitor: Monitor) -> web.Application

### Description
Initializes the aiohttp-based web application.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
None

### Response
#### Success Response (200)
- **application** (web.Application) - The initialized aiohttp web application.

### Response Example
None
```

--------------------------------

### Show Task Creation Stack with 'where' command

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Use the `where` command followed by a task ID to display the task creation stack chain, showing how the task was initiated.

```bash
monitor >>> where 139823456782
```

--------------------------------

### Monitor CLI Usage

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/termui.md

This is the general usage syntax for the monitor CLI. To see the usage of each command, run them with the "--help" option.

```shell
monitor_cli [OPTIONS] COMMAND [ARGS]...
```

--------------------------------

### Connect to Aiomonitor CLI

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Connect to a running aiomonitor instance using either the `aiomonitor.cli` module or `telnet`. Ensure you use the correct host and port.

```bash
# Connect to running monitor
$ python -m aiomonitor.cli -H 127.0.0.1 -p 20101
# or
$ telnet localhost 20101

Asyncio Monitor: 3 tasks running
Type help for available commands
monitor >>>
```

--------------------------------

### Run aiomonitor CLI Client

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Use `python -m aiomonitor.cli` to connect to a running aiomonitor instance. Specify host and port for remote connections. The `--help` flag provides usage details.

```bash
# Default connection (127.0.0.1:20101)
$ python -m aiomonitor.cli

# Connect to a remote host and custom port
$ python -m aiomonitor.cli -H 192.168.1.100 -p 20101

# Full help
$ python -m aiomonitor.cli --help
# usage: cli.py [-h] [-H MONITOR_HOST] [-p MONITOR_PORT]
# options:
#   -H, --host  monitor host ip  (default: 127.0.0.1)
#   -p, --port  monitor port number  (default: 20101)

```

--------------------------------

### aiomonitor Help Command

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/tutorial.md

Displays the available commands and their usage within the aiomonitor console.

```bash
monitor >>> help
Usage: help [OPTIONS] COMMAND [ARGS]...

  To see the usage of each command, run them with "--help" option.

Commands:
  cancel (ca)             Cancel an indicated task
  console                 Switch to async Python REPL
  exit (q,quit)           Leave the monitor client session
  help (?,h)              Show the list of commands
  ps (p)                  Show task table
  ps-terminated (pst,pt)  List recently terminated/cancelled tasks
  signal                  Send a Unix signal
  stacktrace (st,stack)   Print a stack trace from the event loop thread
  where (w)               Show stack frames and the task creation chain of a task
  where-terminated (wt)   Show stack frames and the termination/cancellation chain of a task

```

--------------------------------

### Integrate aiomonitor with aiohttp

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Wrap `web._run_app` within `aiomonitor.start_monitor` to enable live introspection of aiohttp applications. Set `hook_task_factory=True` to trace HTTP handler tasks and use `locals` to expose application objects to the monitor's REPL.

```python
import asyncio
import aiomonitor
from aiohttp import web

async def slow_handler(request: web.Request) -> web.Response:
    # This task will appear in `ps` and `where` commands during the sleep
    await asyncio.sleep(30)
    return web.Response(text="done")

async def health(request: web.Request) -> web.Response:
    return web.Response(text="ok")

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

    loop = asyncio.get_running_loop()
    with aiomonitor.start_monitor(
        loop,
        hook_task_factory=True,
        locals={"app": app},
    ):
        # While server runs, connect via telnet to inspect active request tasks
        await web._run_app(app, host="localhost", port=8090)

# $ python app.py
# ======== Running on http://localhost:8090 ======== 
# Monitor running at telnet://127.0.0.1:20101

# In another terminal, trigger a slow request then inspect it:
# $ curl http://localhost:8090/slow &
# $ python -m aiomonitor.cli
# monitor >>> ps
# Task ID         State    Coroutine      Created Location       Since
# 139823456790    PENDING  slow_handler   aiohttp/web_proto:123  0:00:03
# monitor >>> where 139823456790

asyncio.run(main())
```

--------------------------------

### Enter Python REPL in aiomonitor

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/tutorial.md

Command to enter the interactive Python REPL within the running aiomonitor session. Allows for debugging and state exploration.

```bash
monitor >>> console
Python 3.11.7 (main, Dec  9 2023, 21:41:50) [GCC 11.4.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
---
This console is running in an asyncio event loop.
It allows you to wait for coroutines using the 'await' syntax.
Try: await asyncio.sleep(1, result=3)
---
>>>

```

--------------------------------

### monitor_cli

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/termui.md

Command-line interface for aiomonitor.

```APIDOC
## monitor_cli [OPTIONS] COMMAND [ARGS]...

### Description
Command-line interface for aiomonitor. To see the usage of each command, run them with “–help” option.

### Commands
* **cancel**: Cancel an indicated task
* **console**: Switch to async Python REPL
* **exit**: Leave the monitor client session
* **help**: Show the list of commands
* **ps**: Show task table
* **ps-terminated**: List recently terminated/cancelled tasks
* **signal**: Send a Unix signal
* **stacktrace**: Print a stack trace from the event loop
* **where**: Show stack frames and the task creation
* **where-terminated**: Show stack frames and the...
```

--------------------------------

### aiomonitor.termui.commands.custom_help_option

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/termui.md

A custom help option to ensure setting command_done_event.

```APIDOC
## custom_help_option(cmdfunc)

### Description
A custom help option to ensure setting command_done_event.

### Parameters
* **cmdfunc** (callable) - The command function to process.
```

--------------------------------

### Connect to aiomonitor using aiomonitor CLI

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/tutorial.md

Alternative method to connect to aiomonitor using its built-in CLI client. Useful if a standard telnet client is unavailable.

```bash
$ python -m aiomonitor.cli
Asyncio Monitor: 1 tasks running
Type help for commands
monitor >>>

```

--------------------------------

### Run the Test Suite

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/contributing.md

Execute the full test suite with code coverage reporting. This command runs pytest and measures code coverage.

```bash
python -m pytest --cov tests
```

--------------------------------

### List Running Tasks with 'ps' command

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Use the `ps` command in the aiomonitor shell to list all currently running tasks. Options like `--filter` can be used to narrow down the results.

```bash
monitor >>> ps
3 tasks running
Task ID         State    Name      Coroutine       Created Location    Since
139823456781    PENDING  Task-1    worker          app.py:5            0:00:12
139823456782    PENDING  Task-2    db_poller       app.py:8            0:00:12
139823456783    PENDING  Task-3    http_server     app.py:11           0:00:12

monitor >>> ps --filter worker
1 tasks running (showing 1 tasks)
```

--------------------------------

### Connect to aiomonitor via Telnet

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/tutorial.md

Connect to the running aiomonitor instance using a telnet client. Requires the server to be running on port 20101.

```bash
$ telnet localhost 20101
Asyncio Monitor: 1 tasks running
Type help for commands
monitor >>>

```

--------------------------------

### aiomonitor.webui.app.show_about_page

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/webui.md

Displays the 'About' page of the web UI.

```APIDOC
## async show_about_page(request: Request) -> Response

### Description
Displays the 'About' page, which typically contains information about the application.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
None

### Response
#### Success Response (200)
- **response** (Response) - An HTTP response object representing the 'About' page.

### Response Example
None
```

--------------------------------

### aiomonitor.webui.app.show_list_page

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/webui.md

Displays the main list page of the web UI, likely showing tasks.

```APIDOC
## async show_list_page(request: Request) -> Response

### Description
Displays the main list page, which could show running or terminated tasks.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
None

### Response
#### Success Response (200)
- **response** (Response) - An HTTP response object representing the list page.

### Response Example
None
```

--------------------------------

### aiomonitor.termui.completion.ClickCompleter

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/termui.md

Auto-completion helper for Click-based CLIs.

```APIDOC
## class ClickCompleter(root_command: Command | Group)

### Description
Auto-completion helper for Click-based CLIs. You may inspect or modify the auto-completion helpers here.

#### Method: get_completions

##### Description
This should be a generator that yields `Completion` instances. If the generation of completions is something expensive (that takes a lot of time), consider wrapping this Completer class in a ThreadedCompleter. In that case, the completer algorithm runs in a background thread and completions will be displayed as soon as they arrive.

##### Parameters
* **document** (Document) - `Document` instance.
* **complete_event** (CompleteEvent) - `CompleteEvent` instance.
```

--------------------------------

### monitor.console_locals

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

A dictionary of variables that are directly accessible in the `console` REPL. Can be populated via `start_monitor(locals=...)` or mutated at runtime via `monitor.console_locals`.

```APIDOC
## monitor.console_locals

### Description
A dictionary of variables that are directly accessible in the `console` REPL. Can be populated via `start_monitor(locals=...)` or mutated at runtime via `monitor.console_locals`.

### Usage
Variables can be added to `monitor.console_locals` before or after the monitor has started.

### Example
```python
import asyncio
import aiomonitor

async def main():
    loop = asyncio.get_running_loop()
    db_connection = {"host": "db.example.com", "status": "connected"}
    cache = {"hits": 0, "misses": 0}

    with aiomonitor.start_monitor(loop, locals={"db": db_connection}) as monitor:
        # Add more variables after start
        monitor.console_locals["cache"] = cache
        monitor.console_locals["loop"] = loop

        print("In the monitor console, try:")
        print("  >>> db")
        print("  >>> cache['hits']")
        print("  >>> await asyncio.sleep(1, result='hello')")

        while True:
            await asyncio.sleep(60)

# Console session example:
# monitor >>> console
# >>> db
# {'host': 'db.example.com', 'status': 'connected'}
# >>> cache
# {'hits': 0, 'misses': 0}
# >>> exit()
# ✓ The console session is closed.
```
```

--------------------------------

### Enter Async Python REPL with 'console' command

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Use the `console` command to drop into an asynchronous Python REPL within the context of the running event loop. You can interact with tasks and other asyncio objects. Type `exit()` to close the console session.

```bash
monitor >>> console
>>> tasks = [t for t in __import__('asyncio').all_tasks()]
>>> len(tasks)
3
>>> exit()
✓ The console session is closed.
```

--------------------------------

### Expose variables to the aiomonitor REPL console

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Populate `monitor.console_locals` with variables accessible in the async Python REPL. This can be done during initialization or mutated at runtime.

```python
import asyncio
import aiomonitor

async def main():
    loop = asyncio.get_running_loop()
    db_connection = {"host": "db.example.com", "status": "connected"}
    cache = {"hits": 0, "misses": 0}

    with aiomonitor.start_monitor(loop, locals={"db": db_connection}) as monitor:
        # Add more variables after start
        monitor.console_locals["cache"] = cache
        monitor.console_locals["loop"] = loop

        print("In the monitor console, try:")
        print("  >>> db")
        print("  >>> cache['hits']")
        print("  >>> await asyncio.sleep(1, result='hello')")

        while True:
            await asyncio.sleep(60)

# Console session example:
# monitor >>> console
# >>> db
# {'host': 'db.example.com', 'status': 'connected'}
# >>> cache
# {'hits': 0, 'misses': 0}
# >>> exit()
# ✓ The console session is closed.
```

--------------------------------

### Show Termination/Cancellation Stack with 'where-terminated' command

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Use the `where-terminated` command followed by a trace ID to view the termination or cancellation stack trace for a task.

```bash
monitor >>> where-terminated AABB1122
```

--------------------------------

### aiomonitor.webui.app.get_live_task_list

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/webui.md

Retrieves a list of currently running tasks.

```APIDOC
## async get_live_task_list(request: Request) -> Response

### Description
Fetches and returns a list of all tasks that are currently running.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
None

### Response
#### Success Response (200)
- **tasks** (list) - A list of running tasks, where each task is represented by a dictionary or object.

### Response Example
None
```

--------------------------------

### Exiting the Console

Source: https://github.com/aio-libs/aiomonitor/blob/main/README.rst

Shows how to exit the aiomonitor console session by typing 'exit()' or pressing Ctrl+D.

```python
>>> exit()

    ✓ The console session is closed.
    monitor >>>
```

--------------------------------

### Enable Task Creation and Cancellation Chain Tracking with hook_task_factory

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Enable aiomonitor to replace the event loop's task factory, wrapping all tasks as `TracedTask`. This allows `where` and `where-terminated` commands to display full recursive task creation stacks and cancellation chains. Ensure `aiomonitor` is imported.

```python
import asyncio
import aiomonitor

async def level3():
    await asyncio.sleep(100)  # long-running leaf task

async def level2():
    asyncio.create_task(level3(), name="leaf-task")
    await asyncio.sleep(100)

async def level1():
    asyncio.create_task(level2(), name="mid-task")
    await asyncio.sleep(100)

async def main():
    loop = asyncio.get_running_loop()
    with aiomonitor.start_monitor(loop, hook_task_factory=True):
        asyncio.create_task(level1(), name="root-task")
        await asyncio.sleep(3600)

# monitor >>> ps
# 4 tasks running
# Task ID          State    Name        Coroutine  Created Location     Since
# 140234567890     PENDING  root-task   level1     app.py:14            0:00:02
# 140234567891     PENDING  mid-task    level2     app.py:9             0:00:02
# 140234567892     PENDING  leaf-task   level3     app.py:5             0:00:02

# monitor >>> where 140234567892
# Stack of the root task or coroutine scheduled in the event loop (most recent call last)
#   File "app.py", line 14, in main
# Stack of Task-mid-task when creating the next task (most recent call last)
#   File "app.py", line 9, in level2
# Stack of Task-leaf-task (most recent call last)
#   File "app.py", line 5, in level3
asyncio.run(main())
```

--------------------------------

### aiomonitor.termui.commands.auto_async_command_done

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/termui.md

Helper function for auto-completion of async commands.

```APIDOC
## auto_async_command_done(cmdfunc)

### Description
Helper function for auto-completion of async commands.

### Parameters
* **cmdfunc** (callable) - The command function to process.
```

--------------------------------

### Expose Local Variables in Python REPL

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/tutorial.md

Configure aiomonitor to expose local variables in the Python REPL by passing a 'locals' dictionary to start_monitor.

```python
locals = {"foo": "bar"}
with aiomonitor.start_monitor(loop, locals=locals):
    web.run_app(app, port=20101, host='127.0.0.1')

```

--------------------------------

### aiomonitor.webui.app.show_trace_page

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/webui.md

Displays the trace page of the web UI, likely for debugging.

```APIDOC
## async show_trace_page(request: Request) -> Response

### Description
Displays the trace page, potentially showing detailed execution traces or logs.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
None

### Response
#### Success Response (200)
- **response** (Response) - An HTTP response object representing the trace page.

### Response Example
None
```

--------------------------------

### Custom Help Option Decorator

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/termui.md

Use this decorator to ensure the command_done_event is set for custom help options in commands.

```python
def custom_help_option(cmdfunc):
    """A custom help option to ensure setting command_done_event."""
    return cmdfunc
```

--------------------------------

### Run Individual Tests

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/contributing.md

Run specific tests within a file, with verbose output and filtering by test name. Useful for targeted debugging.

```bash
python -m pytest -sv tests/test_monitor.py -k test_name
```

--------------------------------

### Register Custom CLI Commands with aiomonitor

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Extend the aiomonitor shell by registering new commands using `@monitor_cli.command()`. Use `@auto_command_done` for synchronous and `@auto_async_command_done` for asynchronous commands. `@custom_help_option` provides a compatible help flag.

```python
import asyncio
import click
import aiomonitor
from aiomonitor.termui.commands import (
    auto_async_command_done,
    auto_command_done,
    custom_help_option,
    print_ok,
    print_fail,
)

# Synchronous custom command
@aiomonitor.monitor_cli.command(name="status")
@click.argument("component", default="all")
@custom_help_option
@auto_command_done
def do_status(ctx: click.Context, component: str) -> None:
    """Show application component status."""
    monitor = ctx.obj
    components = monitor.console_locals.get("components", {})
    if component == "all":
        for name, state in components.items():
            click.echo(f"  {name}: {state}")
    elif component in components:
        click.echo(f"{component}: {components[component]}")
    else:
        print_fail(f"Unknown component: {component}")

# Async custom command
@aiomonitor.monitor_cli.command(name="fetch")
@click.argument("url")
@custom_help_option
def do_fetch(ctx: click.Context, url: str) -> None:
    """Fetch a URL and print status code."""
    import aiohttp

    @auto_async_command_done
    async def _fetch(ctx):
        async with aiohttp.ClientSession() as sess:
            async with sess.get(url) as resp:
                print_ok(f"GET {url} -> {resp.status}")

    asyncio.create_task(_fetch(ctx))


async def main():
    loop = asyncio.get_running_loop()
    components = {"db": "running", "cache": "running", "queue": "paused"}
    with aiomonitor.start_monitor(loop, locals={"components": components}):
        while True:
            await asyncio.sleep(60)

# monitor >>> status
#   db: running
#   cache: running
#   queue: paused
# monitor >>> fetch https://httpbin.org/get
# ✓ GET https://httpbin.org/get -> 200

asyncio.run(main())

```

--------------------------------

### Accessing Exposed Local Variable

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/tutorial.md

Demonstrates accessing an exposed local variable ('foo') within the aiomonitor Python REPL.

```bash
monitor >>> console
>>> foo
bar
>>> exit()
monitor >>>

```

--------------------------------

### Programmatic Usage of aiomonitor CLI Client

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Connect to aiomonitor programmatically using `monitor_client` (blocking) or `async_monitor_client` (async). The `TelnetClient` offers low-level control for interaction.

```python
# Programmatic usage of the CLI client
import asyncio
from aiomonitor.cli import monitor_client, async_monitor_client
from aiomonitor.telnet import TelnetClient

# Blocking version
monitor_client("127.0.0.1", 20101)

# Async version
async def connect():
    await async_monitor_client("127.0.0.1", 20101)

# Low-level TelnetClient
async def raw_connect():
    async with TelnetClient("127.0.0.1", 20101) as client:
        await client.interact()

asyncio.run(raw_connect())

```

--------------------------------

### Exit Python REPL

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/tutorial.md

Command to exit the Python REPL and return to the aiomonitor prompt.

```python
>>> exit()
monitor >>>

```

--------------------------------

### Dump Event Loop Stacktrace with 'stacktrace' command

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Use the `stacktrace` command to dump the current stack trace of the event loop thread. This is useful for debugging.

```bash
monitor >>> stacktrace
```

--------------------------------

### aiomonitor.termui.commands.auto_command_done

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/termui.md

Helper function for auto-completion of commands.

```APIDOC
## auto_command_done(cmdfunc)

### Description
Helper function for auto-completion of commands.

### Parameters
* **cmdfunc** (callable) - The command function to process.
```

--------------------------------

### Click Completer Class

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/termui.md

Implement this class to create custom auto-completion logic for the CLI. Consider using ThreadedCompleter for expensive completion operations.

```python
class ClickCompleter:
    """Root command for auto-completion."""
    def __init__(self, root_command: Command | Group):
        ...

    def get_completions(self, document: Document, complete_event: CompleteEvent) -> Iterable[Completion]:
        """This should be a generator that yields `Completion` instances."""
        ...
        # If the generation of completions is something expensive (that takes a
        # lot of time), consider wrapping this Completer class in a
        # ThreadedCompleter. In that case, the completer algorithm runs in a
        # background thread and completions will be displayed as soon as they
        # arrive.
        # 
        # * **Parameters:**
        #   * **document** – `Document` instance.
        #   * **complete_event** – `CompleteEvent` instance.
```

--------------------------------

### aiohttp Application with Monitor

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/examples.md

Integrates aiomonitor into an aiohttp web application, enabling monitoring of tasks and application state. `hook_task_factory=True` is used to monitor all tasks created within the application.

```python
import asyncio
import logging

import uvloop
from aiohttp import web

import aiomonitor


async def inner2() -> None:
    await asyncio.sleep(100)


async def inner1(tg: asyncio.TaskGroup) -> None:
    t = tg.create_task(inner2())
    await t


async def simple(request: web.Request) -> web.Response:
    print("Start sleeping")
    async with asyncio.TaskGroup() as tg:
        tg.create_task(inner1(tg))
    print("Finished sleeping")
    return web.Response(text="Simple answer")


async def main() -> None:
    loop = asyncio.get_running_loop()
    app = web.Application()
    app.router.add_get("/simple", simple)
    with aiomonitor.start_monitor(loop, hook_task_factory=True):
        await web._run_app(app, port=8090, host="localhost")


if __name__ == "__main__":
    logging.basicConfig()
    logging.getLogger("aiomonitor").setLevel(logging.DEBUG)
    uvloop.install()
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass
```

--------------------------------

### Complete Signal Names Helper

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/termui.md

A helper function to provide auto-completion for signal names in the CLI.

```python
def complete_signal_names(ctx: Context, param: Parameter, incomplete: str) -> Iterable[str]:
    ...
```

--------------------------------

### Send Unix Signal with 'signal' command

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Use the `signal` command to send a Unix signal to the running process. Specify the signal name (e.g., `SIGUSR1`) and the command confirms the action.

```bash
monitor >>> signal SIGUSR1
✓ Sent signal to SIGUSR1 PID 12345
```

--------------------------------

### aiomonitor.termui.commands.interact

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/termui.md

The interactive loop for each telnet client connection.

```APIDOC
## async interact(self: Monitor, connection: TelnetConnection) -> None

### Description
The interactive loop for each telnet client connection.

### Parameters
* **self** (Monitor) - The monitor instance.
* **connection** (TelnetConnection) - The telnet connection object.
```

--------------------------------

### aiomonitor.webui.app.get_version

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/webui.md

Retrieves the current version of the aiomonitor.

```APIDOC
## async get_version(request: Request) -> Response

### Description
Returns the current version of the aiomonitor.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
None

### Response
#### Success Response (200)
- **version** (str) - The version string of aiomonitor.

### Response Example
None
```

--------------------------------

### aiomonitor.termui.completion.complete_signal_names

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/termui.md

Provides auto-completion for signal names.

```APIDOC
## complete_signal_names(ctx: Context, param: Parameter, incomplete: str) -> Iterable[str]

### Description
Provides auto-completion for signal names.

### Parameters
* **ctx** (Context) - The Click context.
* **param** (Parameter) - The parameter being completed.
* **incomplete** (str) - The incomplete input string.
```

--------------------------------

### Complete Trace ID Helper

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/termui.md

A helper function to provide auto-completion for trace IDs in the CLI.

```python
def complete_trace_id(ctx: Context, param: Parameter, incomplete: str) -> Iterable[str]:
    ...
```

--------------------------------

### Complete Task ID Helper

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/termui.md

A helper function to provide auto-completion for task IDs in the CLI.

```python
def complete_task_id(ctx: Context, param: Parameter, incomplete: str) -> Iterable[str]:
    ...
```

--------------------------------

### aiomonitor.termui.completion.complete_task_id

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/termui.md

Provides auto-completion for task IDs.

```APIDOC
## complete_task_id(ctx: Context, param: Parameter, incomplete: str) -> Iterable[str]

### Description
Provides auto-completion for task IDs.

### Parameters
* **ctx** (Context) - The Click context.
* **param** (Parameter) - The parameter being completed.
* **incomplete** (str) - The incomplete input string.
```

--------------------------------

### aiomonitor.termui.completion.complete_trace_id

Source: https://github.com/aio-libs/aiomonitor/blob/main/docs/reference/termui.md

Provides auto-completion for trace IDs.

```APIDOC
## complete_trace_id(ctx: Context, param: Parameter, incomplete: str) -> Iterable[str]

### Description
Provides auto-completion for trace IDs.

### Parameters
* **ctx** (Context) - The Click context.
* **param** (Parameter) - The parameter being completed.
* **incomplete** (str) - The incomplete input string.
```

--------------------------------

### List Terminated Tasks with 'ps-terminated' command

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Use the `ps-terminated` command to view a list of recently terminated tasks. This command may strip older entries if `max_termination_history` is exceeded.

```bash
monitor >>> ps-terminated
5 tasks terminated (old ones may be stripped)
Trace ID    Name      Coro        Since Started  Since Terminated
AABB1122    Task-10   job         0:00:30        0:00:05
```

--------------------------------

### Cancel a Task with 'cancel' command

Source: https://context7.com/aio-libs/aiomonitor/llms.txt

Use the `cancel` command followed by a task ID to cancel a running task. A confirmation message is returned upon successful cancellation.

```bash
monitor >>> cancel 139823456782
✓ Cancelled task 139823456782
```