Skip to main content

​
Overview

This page showcases real addon examples from examples/addons/ in the mitmproxy source code. Each example demonstrates practical use cases and best practices.
All examples can be run with: mitmproxy -s examples/addons/example-name.py

​
Basic Examples

​
Simple Request Counter

The most basic addon - counting requests as they flow through the proxy.
examples/addons/anatomy.py
import logging

class Counter:
    def __init__(self):
        self.num = 0

    def request(self, flow):
        self.num = self.num + 1
        logging.info("We've seen %d flows" % self.num)

addons = [Counter()]
Usage: 
mitmproxy -s anatomy.py

​
Adding HTTP Headers

Add a custom header to every HTTP response.
examples/addons/http-add-header.py
class AddHeader:
    def __init__(self):
        self.num = 0

    def response(self, flow):
        self.num = self.num + 1
        flow.response.headers["count"] = str(self.num)

addons = [AddHeader()]
Key concepts:
  • Access response headers via flow.response.headers
  • Headers are case-insensitive dictionaries
  • State persists across requests

​
HTTP Modification

​
Modifying Query Parameters

Add or modify URL query parameters on the fly.
examples/addons/http-modify-query-string.py
from mitmproxy import http

def request(flow: http.HTTPFlow) -> None:
    flow.request.query["mitmproxy"] = "rocks"
You can define hooks as module-level functions instead of class methods. Both styles work.

​
Modifying Form Data

Intercept and modify POST form data.
examples/addons/http-modify-form.py
from mitmproxy import http

def request(flow: http.HTTPFlow) -> None:
    # Modify form data
    if flow.request.urlencoded_form:
        flow.request.urlencoded_form["username"] = "modified_user"

​
Redirecting Requests

Redirect specific requests to different URLs.
examples/addons/http-redirect-requests.py
from mitmproxy import http

def request(flow: http.HTTPFlow) -> None:
    if flow.request.pretty_host == "example.com":
        flow.request.host = "example.org"

​
Replying from Proxy

Generate responses directly from the proxy without contacting the server.
examples/addons/http-reply-from-proxy.py
from mitmproxy import http

def request(flow: http.HTTPFlow) -> None:
    if flow.request.path == "/local":
        flow.response = http.Response.make(
            200,
            b"Hello from mitmproxy!",
            {"Content-Type": "text/plain"}
        )
Use cases:
  • Mock API endpoints
  • Block requests
  • Serve local content

​
Streaming

​
Enabling Response Streaming

Stream large responses to save memory.
examples/addons/http-stream-simple.py
def responseheaders(flow):
    """
    Enables streaming for all responses.
    Equivalent to --set stream_large_bodies=1
    """
    flow.response.stream = True
Enable streaming in the responseheaders hook before the body is received.

​
Conditional Streaming

Stream only large responses.
examples/addons/http-stream-modify.py
def responseheaders(flow):
    content_length = flow.response.headers.get("content-length", 0)
    if int(content_length) > 1000000:  # 1MB
        flow.response.stream = True

​
Options and Configuration

​
Adding Custom Options

Create addons with configurable options.
examples/addons/options-simple.py
from mitmproxy import ctx

class AddHeader:
    def __init__(self):
        self.num = 0

    def load(self, loader):
        loader.add_option(
            name="addheader",
            typespec=bool,
            default=False,
            help="Add a count header to responses",
        )

    def response(self, flow):
        if ctx.options.addheader:
            self.num = self.num + 1
            flow.response.headers["count"] = str(self.num)

addons = [AddHeader()]
Usage: 
mitmproxy -s options-simple.py --set addheader=true

​
Reacting to Option Changes

Handle configuration changes at runtime.
examples/addons/options-configure.py
from mitmproxy import ctx
from mitmproxy.addonmanager import Loader
import logging

class OptionExample:
    def load(self, loader: Loader):
        loader.add_option(
            name="my_threshold",
            typespec=int,
            default=100,
            help="Response size threshold"
        )

    def configure(self, updated):
        if "my_threshold" in updated:
            logging.info(f"Threshold changed to: {ctx.options.my_threshold}")

    def response(self, flow):
        if len(flow.response.content) > ctx.options.my_threshold:
            logging.info(f"Large response: {len(flow.response.content)} bytes")

addons = [OptionExample()]

​
Filtering

​
Flow Filtering

Use mitmproxy’s powerful filter syntax in addons.
examples/addons/filter-flows.py
import logging
from mitmproxy import flowfilter, http
from mitmproxy.addonmanager import Loader

class Filter:
    filter: flowfilter.TFilter

    def load(self, loader: Loader):
        loader.add_option(
            "flowfilter",
            str,
            "",
            "Check that flow matches filter."
        )

    def configure(self, updated):
        if "flowfilter" in updated:
            self.filter = flowfilter.parse(".")

    def response(self, flow: http.HTTPFlow) -> None:
        if flowfilter.match(self.filter, flow):
            logging.info("Flow matches filter:")
            logging.info(flow)

addons = [Filter()]
Filter examples:
  • ~d example.com - Domain matches
  • ~m POST - Method matches
  • ~c 200 - Status code matches
  • ~u /api/ - URL path contains

​
Commands

​
Simple Command

Add custom commands to mitmproxy’s command prompt.
examples/addons/commands-simple.py
import logging
from mitmproxy import command

class MyAddon:
    def __init__(self):
        self.num = 0

    @command.command("myaddon.inc")
    def inc(self) -> None:
        self.num += 1
        logging.info(f"num = {self.num}")

addons = [MyAddon()]
Usage in mitmproxy: 
: myaddon.inc

​
Commands with Flow Arguments

Create commands that operate on flows.
examples/addons/commands-flows.py
import logging
from mitmproxy import command, flow

class MyAddon:
    @command.command("myaddon.duplicate")
    def duplicate(self, flows: list[flow.Flow]) -> None:
        for f in flows:
            logging.info(f"Duplicating: {f.request.url}")
            # Create duplicate

addons = [MyAddon()]

​
WebSocket

​
WebSocket Message Processing

Intercept and modify WebSocket messages.
examples/addons/websocket-simple.py
import logging
import re
from mitmproxy import http

def websocket_message(flow: http.HTTPFlow):
    assert flow.websocket is not None
    message = flow.websocket.messages[-1]

    # Check message direction
    if message.from_client:
        logging.info(f"Client sent: {message.content!r}")
    else:
        logging.info(f"Server sent: {message.content!r}")

    # Modify message content
    message.content = re.sub(rb"^Hello", b"HAPPY", message.content)

    # Drop messages
    if b"FOOBAR" in message.content:
        message.drop()
Key concepts:
  • Access messages via flow.websocket.messages[-1]
  • Check direction with message.from_client
  • Drop messages with message.drop()

​
WebSocket Message Injection

Inject custom messages into WebSocket connections.
examples/addons/websocket-inject-message.py
import logging
from mitmproxy import ctx, http
from mitmproxy.websocket import WebSocketMessage

class InjectWebSocket:
    def websocket_start(self, flow: http.HTTPFlow) -> None:
        # Store flow for later injection
        self.flow = flow

    def inject_message(self):
        if hasattr(self, 'flow') and self.flow.websocket:
            msg = WebSocketMessage(
                type=0x1,  # Text message
                from_client=False,
                content=b"Injected message"
            )
            self.flow.websocket.messages.append(msg)

addons = [InjectWebSocket()]

​
TCP

​
TCP Message Processing

Handle raw TCP connections.
examples/addons/tcp-simple.py
import logging
from mitmproxy import tcp
from mitmproxy.utils import strutils

def tcp_message(flow: tcp.TCPFlow):
    message = flow.messages[-1]
    message.content = message.content.replace(b"foo", b"bar")

    logging.info(
        f"tcp_message[from_client={message.from_client}, "
        f"content={strutils.bytes_to_escaped_str(message.content)}]"
    )
TCP is stream-based. Message boundaries are arbitrary and based on socket.recv() calls.
Enable TCP interception: 
mitmdump --tcp-hosts ".*" -s tcp-simple.py

​
DNS

​
DNS Response Spoofing

Modify or spoof DNS responses.
examples/addons/dns-simple.py
import ipaddress
import logging
from mitmproxy import dns

def dns_request(flow: dns.DNSFlow) -> None:
    q = flow.request.question
    if q and q.type == dns.types.AAAA:
        logging.info(f"Spoofing IPv6 records for {q.name}...")
        
        if q.name == "example.com":
            flow.response = flow.request.succeed([
                dns.ResourceRecord(
                    name="example.com",
                    type=dns.types.AAAA,
                    class_=dns.classes.IN,
                    ttl=dns.ResourceRecord.DEFAULT_TTL,
                    data=ipaddress.ip_address("::1").packed,
                )
            ])
        elif q.name == "example.org":
            flow.response = flow.request.fail(dns.response_codes.NXDOMAIN)
        else:
            flow.response = flow.request.succeed([])
Use cases:
  • Local DNS resolution
  • Blocking domains
  • Testing DNS failures

​
Async Operations

​
Non-blocking Addon

Use async/await for non-blocking I/O.
examples/addons/nonblocking.py
import asyncio
import logging

# Async hooks (preferred)
async def request(flow):
    logging.info(f"Handle request: {flow.request.host}{flow.request.path}")
    await asyncio.sleep(5)  # Non-blocking delay
    logging.info(f"Start request: {flow.request.host}{flow.request.path}")
Alternative with threads: 
from mitmproxy.script import concurrent
import time

@concurrent
def request(flow):
    logging.info(f"Handle request: {flow.request.host}{flow.request.path}")
    time.sleep(5)  # Runs in separate thread
    logging.info(f"Start request: {flow.request.host}{flow.request.path}")
Async hooks are preferred over @concurrent as they provide better performance without race conditions.

​
Advanced Examples

​
Duplicate, Modify, and Replay

Duplicate flows, modify them, and replay to the server.
examples/addons/duplicate-modify-replay.py
import logging
from mitmproxy import ctx, http

class DuplicateAndReplay:
    def response(self, flow: http.HTTPFlow) -> None:
        if flow.request.path == "/api/data":
            # Create a duplicate
            duplicate = flow.copy()
            duplicate.request.path = "/api/data/v2"
            duplicate.request.headers["X-Replayed"] = "true"
            
            # Replay the modified request
            ctx.master.commands.call("replay.client", [duplicate])
            logging.info(f"Replayed modified request to {duplicate.request.url}")

addons = [DuplicateAndReplay()]

​
Flow Persistence

Save and load flows from disk.
examples/addons/io-write-flow-file.py
import logging
from mitmproxy import ctx, http, io

class FlowWriter:
    def __init__(self):
        self.f = None

    def load(self, loader):
        loader.add_option(
            "flowfile",
            str,
            "flows.mitm",
            "File to save flows"
        )

    def configure(self, updates):
        if "flowfile" in updates:
            if self.f:
                self.f.close()
            self.f = open(ctx.options.flowfile, "wb")
            self.w = io.FlowWriter(self.f)

    def response(self, flow: http.HTTPFlow):
        if self.w:
            self.w.add(flow)

    def done(self):
        if self.f:
            self.f.close()

addons = [FlowWriter()]

​
Reading Saved Flows

examples/addons/io-read-saved-flows.py
import logging
from mitmproxy import io

class FlowReader:
    def load(self, loader):
        loader.add_option(
            "flowfile",
            str,
            "",
            "Flow file to read"
        )

    def running(self):
        if ctx.options.flowfile:
            with open(ctx.options.flowfile, "rb") as f:
                reader = io.FlowReader(f)
                for flow in reader.stream():
                    logging.info(f"Loaded: {flow.request.url}")

addons = [FlowReader()]

​
HTTP Trailers

Handle HTTP trailers (rare but useful for some protocols).
examples/addons/http-trailers.py
import logging
from mitmproxy import http

def request(flow: http.HTTPFlow) -> None:
    # Add request trailers
    if not flow.request.trailers:
        flow.request.trailers = http.Headers()
    flow.request.trailers["X-Request-Trailer"] = "value"

def response(flow: http.HTTPFlow) -> None:
    # Read response trailers
    if flow.response.trailers:
        logging.info(f"Response trailers: {flow.response.trailers}")

​
Testing and Debugging

​
Logging Events

Log all events for debugging.
examples/addons/log-events.py
import logging

class LogEvents:
    def request(self, flow):
        logging.info(f"Request: {flow.request.url}")

    def response(self, flow):
        logging.info(f"Response: {flow.response.status_code}")

    def error(self, flow):
        logging.error(f"Error: {flow.error.msg}")

    def websocket_message(self, flow):
        message = flow.websocket.messages[-1]
        logging.info(f"WebSocket: {message.content[:50]}")

    def tcp_message(self, flow):
        message = flow.messages[-1]
        logging.info(f"TCP: {len(message.content)} bytes")

addons = [LogEvents()]

​
Shutdown Handling

Perform cleanup on shutdown.
examples/addons/shutdown.py
import logging

class ShutdownExample:
    def running(self):
        logging.info("Addon started and running")

    def done(self):
        logging.info("Addon shutting down")
        # Cleanup resources
        # Close connections
        # Save state

addons = [ShutdownExample()]

​
Integration Examples

​
WSGI/Flask Integration

Serve a Flask app through mitmproxy.
examples/addons/wsgi-flask-app.py
from flask import Flask
from mitmproxy import ctx, http

app = Flask("proxapp")

@app.route("/")
def hello():
    return "Hello from Flask!"

class WSGIApp:
    def request(self, flow: http.HTTPFlow) -> None:
        if flow.request.host == "proxapp":
            flow.response = http.Response.make(
                200,
                b"Flask app response"
            )

addons = [WSGIApp()]

​
Best Practices

Use Logging

Always use logging module instead of print() for output.

Handle Errors

Wrap risky operations in try/except to avoid breaking other addons.

Check Flow State

Always verify that flow.response or flow.websocket exists before accessing.

Document Options

Provide clear help text for custom options.

​
Running Examples

All examples are located in examples/addons/ and can be run with: 
# Run with mitmproxy (interactive)
mitmproxy -s examples/addons/example-name.py

# Run with mitmdump (command-line)
mitmdump -s examples/addons/example-name.py

# Run with mitmweb (web interface)
mitmweb -s examples/addons/example-name.py

# Set options
mitmproxy -s example.py --set option_name=value

​
Next Steps

Event Hooks

Complete reference of all available hooks

Built-in Addons

Learn from production-quality built-in addons

Build docs developers (and LLMs) love

Get started for freeTalk to us