Event Hooks

from mitmproxy.addonmanager import Loader

def load(self, loader: Loader) -> None:
    loader.add_option(
        name="my_option",
        typespec=bool,
        default=False,
        help="Enable my feature"
    )

def configure(self, updated: set[str]) -> None:
    if "my_option" in updated:
        # React to option changes
        pass

def running(self) -> None:
    logging.info("Proxy is ready!")

def done(self) -> None:
    # Cleanup resources
    self.close_connections()

def update(self, flows: Sequence[mitmproxy.flow.Flow]) -> None:
    for flow in flows:
        # React to flow updates
        pass

from mitmproxy import http

def requestheaders(self, flow: http.HTTPFlow) -> None:
    # Modify headers before body is received
    flow.request.headers["X-Custom"] = "value"

def request(self, flow: http.HTTPFlow) -> None:
    # Access complete request including body
    if b"password" in flow.request.content:
        flow.request.content = b"REDACTED"

def responseheaders(self, flow: http.HTTPFlow) -> None:
    # Enable streaming for large responses
    if int(flow.response.headers.get("content-length", 0)) > 1000000:
        flow.response.stream = True

def response(self, flow: http.HTTPFlow) -> None:
    # Modify response
    flow.response.status_code = 200
    flow.response.text = "Modified content"

def error(self, flow: http.HTTPFlow) -> None:
    logging.error(f"HTTP error: {flow.error.msg}")

def http_connect(self, flow: http.HTTPFlow) -> None:
    # Block certain CONNECT requests
    if flow.request.host == "blocked.com":
        flow.response = http.Response.make(
            403, b"Forbidden"
        )

def http_connect_upstream(self, flow: http.HTTPFlow) -> None:
    # Set custom authentication for upstream proxy
    flow.request.headers["Proxy-Authorization"] = "Basic ..."

from mitmproxy import connection

def client_connected(self, client: connection.Client) -> None:
    logging.info(f"Client connected: {client.address}")

def client_disconnected(self, client: connection.Client) -> None:
    logging.info(f"Client disconnected: {client.address}")

def server_connect(self, data: ServerConnectionHookData) -> None:
    # Redirect to different server
    data.server.address = ("other.example.com", 443)

def server_connected(self, data: ServerConnectionHookData) -> None:
    logging.info(f"Connected to {data.server.address}")

def server_disconnected(self, data: ServerConnectionHookData) -> None:
    pass

def server_connect_error(self, data: ServerConnectionHookData) -> None:
    logging.error(f"Failed to connect: {data.server.error}")

def websocket_start(self, flow: http.HTTPFlow) -> None:
    logging.info("WebSocket connection started")

def websocket_message(self, flow: http.HTTPFlow) -> None:
    # Get the latest message
    message = flow.websocket.messages[-1]
    
    # Modify message content
    if message.from_client:
        message.content = message.content.replace(b"foo", b"bar")
    
    # Drop message
    if b"DROP" in message.content:
        message.drop()

def websocket_end(self, flow: http.HTTPFlow) -> None:
    close_code = flow.websocket.close_code
    logging.info(f"WebSocket closed: {close_code}")

from mitmproxy import tcp

def tcp_start(self, flow: tcp.TCPFlow) -> None:
    logging.info(f"TCP connection to {flow.server_conn.address}")

def tcp_message(self, flow: tcp.TCPFlow) -> None:
    message = flow.messages[-1]
    # Modify TCP message
    message.content = message.content.replace(b"foo", b"bar")

def tcp_end(self, flow: tcp.TCPFlow) -> None:
    logging.info(f"TCP connection ended: {len(flow.messages)} messages")

def tcp_error(self, flow: tcp.TCPFlow) -> None:
    logging.error(f"TCP error: {flow.error.msg}")

import ipaddress
from mitmproxy import dns

def dns_request(self, flow: dns.DNSFlow) -> None:
    q = flow.request.question
    if q.name == "example.com" and q.type == dns.types.A:
        # Spoof DNS response
        flow.response = flow.request.succeed([
            dns.ResourceRecord(
                name="example.com",
                type=dns.types.A,
                class_=dns.classes.IN,
                ttl=300,
                data=ipaddress.ip_address("127.0.0.1").packed,
            )
        ])

def dns_response(self, flow: dns.DNSFlow) -> None:
    # Log DNS response
    for answer in flow.response.answers:
        logging.info(f"DNS answer: {answer.name} -> {answer.data}")

def dns_error(self, flow: dns.DNSFlow) -> None:
    logging.error(f"DNS error: {flow.error.msg}")

# Class: HttpRequestHook -> Hook name: http_request
# Class: DnsRequestHook -> Hook name: dns_request
# Class: TcpMessageHook -> Hook name: tcp_message

client_connected
  ↓
requestheaders
  ↓
request
  ↓
server_connect (if needed)
  ↓
server_connected
  ↓
responseheaders
  ↓
response
  ↓
client_disconnected (eventually)