import mitmproxy.http
from yarl import URL

class Interceptor:
    def response(self, flow: mitmproxy.http.HTTPFlow):
        respose_url = URL(flow.request.url)
        if respose_url.path != "/feed":  # drop the flows we dont need
            return

        assert flow.response is not None
        response_text = flow.response.text
        # parsing and extracting
        data = json.loads(response_text)
        print(do_extracting(data))

addons = [
    Interceptor()
]

import uiautomator2 as u2

d = u2.connect() # connect to device
d.app_start("com.android.XXX")
... # and more example on repo openatx/uiautomator2

from mitmproxy.addons import core
from mitmproxy.master import Master
from mitmproxy.options import Options
from mitmproxy.proxy import ProxyConfig, ProxyServer


async def main() -> None:
    options = Options(listen_host="0.0.0.0", listen_port=8080)
    master = Master(options)
    master.server = server = ProxyServer(ProxyConfig(options))

    # TODO: Keeps this line until the major version 7 is released
    master.addons.add(core.Core())

    master.start()
    await master.running()
    loop = asyncio.get_event_loop()
    try:
        # Do thing with MITM
        pass
    finally:
        # Don't use `Master.shutdown` method.
        # It will make the process being unable to exit.
        # And I don't know what is causing this.
        master.should_exit.set()
        server.shutdown()


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

...
import contextlib

...
from mitmproxy.http import HTTPFlow


ResponseHook = Callable[[HTTPFlow], None]


class Interceptor:
    def __init__(self):
        self.response_hook: Optional[ResponseHook] = None

    @contextlib.contextmanager
    def hook_response(self, hook_function: ResponseHook):
        self.response_hook = hook_function
        try:
            yield
        finally:
            self.response_hook = None

    def response(self, flow: HTTPFlow):
        if self.response_hook is not None:
            self.response_hook(flow)


@contextlib.asynccontextmanager
async def intercept(listen_port: int = 8080) -> None:
    options = Options(listen_host="0.0.0.0", listen_port=listen_port)
    master = Master(options)
    master.server = server = ProxyServer(ProxyConfig(options))

    # TODO: Keeps this line until the major version 7 is released
    master.addons.add(core.Core())

    master.start()
    await master.running()

    interceptor = Interceptor()
    master.addons.add(interceptor)
    try:
        yield interceptor
    finally:
        # Don't use `Master.shutdown()`.
        # It will make the process being unable to exit.
        # And I don't know what is causing this.
        master.should_exit.set()
        server.shutdown()

from pathlib import Path

import httpx


CAFILE_PATH = Path("~/.mitmproxy/mitmproxy-ca.pem").expanduser().absolute()


def control(interceptor: Interceptor, proxy_port: int):
    def wait_resposne(flow: HTTPFlow):
        # run in main thread
        print("in sync control")
        print("received response url:", flow.request.url)

    with httpx.Client(
        proxies=f"http://localhost:{proxy_port}", verify=CAFILE_PATH
    ) as client:
        # run in another thread
        with interceptor.hook_response(wait_resposne):
            resp = client.get("https://httpbin.org/get")
            resp.raise_for_status()

        resp = client.get("https://httpbin.org/get")
        resp.raise_for_status()


async def async_control(interceptor: Interceptor, proxy_port: int):
    def wait_resposne(flow: HTTPFlow):
        print("in async control")
        print("received response url:", flow.request.url)

    async with httpx.AsyncClient(
        proxies=f"http://localhost:{proxy_port}",
        verify=CAFILE_PATH,
    ) as client:
        with interceptor.hook_response(wait_resposne):
            resp = await client.get("https://httpbin.org/get")
            resp.raise_for_status()

        resp = await client.get("https://httpbin.org/get")
        resp.raise_for_status()


async def main():
    loop = asyncio.get_event_loop()
    port = 8888
    async with intercept(listen_port=port) as interceptor:
        # run in other thread to avoid blocking main thread
        await loop.run_in_executor(None, control, interceptor, port)

        # async function is invoked normally via await statement
        await async_control(interceptor, port)


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

def configure_remote_proxy(proxy_port):
    raise NotImplementedError


def control(interceptor):
    raise NotImplementedError


async def main():
    port = 8888
    async with intercept(listen_port=port) as interceptor:
        # Run automation script to configure proxy setting of remote device.
        await asyncio.to_thread(configure_remote_proxy, proxy_port=port)
        # Due to the uiautomator2 does not support asyncio
        await asyncio.to_thread(control, interceptor=interceptor)



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