Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

[project]
name = "mitmproxy-mcp"
version = "0.6.1"
version = "0.7.0"
description = "MITM proxy manager with MCP integration"
readme = "README.md"
requires-python = ">=3.12,<3.14"
Expand Down
88 changes: 72 additions & 16 deletions src/mitmproxy_mcp/core/server.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import asyncio
import contextlib
import logging
import os
import sys
Expand Down Expand Up @@ -59,6 +60,9 @@ def __init__(self, dump_file: Optional[str] = None):
self.session_variables = {}
self.dump_file = dump_file
self.cli_upstream_proxy: Optional[str] = None
self.default_port = 8080
self.default_host = "127.0.0.1"
self.auto_start = False

def _get_verify_param(self, verify_override: Optional[bool] = None) -> Any:
if verify_override is not None:
Expand Down Expand Up @@ -225,26 +229,46 @@ async def replay_request(
# Global Controller Instance
controller = MitmController()

mcp = FastMCP("Mitmproxy Manager")

@contextlib.asynccontextmanager
async def _lifespan(_server: FastMCP):
"""Auto-start the proxy on boot when --auto-start / MITMPROXY_AUTO_START is set."""
if getattr(controller, "auto_start", False) and not controller.running:
result = await controller.start(port=controller.default_port, host=controller.default_host)
logger.info("auto_start", result=result)
try:
yield
finally:
if controller.running:
await controller.stop()


mcp = FastMCP("Mitmproxy Manager", lifespan=_lifespan)

# --- MCP Tools ---


@mcp.tool()
async def start_proxy(
port: int = 8080, dump_file: Optional[str] = None, upstream_proxy: Optional[str] = None
port: Optional[int] = None,
dump_file: Optional[str] = None,
upstream_proxy: Optional[str] = None,
) -> str:
"""
Start the mitmproxy instance.
Args:
port: Port to listen on (default 8080)
port: Port to listen on. Omit to use the server's configured default
(--port / MITMPROXY_PORT, else 8080).
dump_file: Optional file path to save raw mitmproxy .flow data.
Prefix with + to append to an existing file.
upstream_proxy: Optional upstream proxy URL (e.g., 'http://user:pass@proxy:port').
"""
try:
return await controller.start(
port=port, dump_file=dump_file, upstream_proxy=upstream_proxy
port=port if port is not None else controller.default_port,
host=controller.default_host,
dump_file=dump_file,
upstream_proxy=upstream_proxy,
)
except Exception as e:
logger.error("proxy_start_failed", error=str(e))
Expand Down Expand Up @@ -344,9 +368,7 @@ async def inspect_flows(
if columns and "id" not in columns:
columns.insert(0, "id")

results = controller.recorder.db.get_by_ids(
ids, columns=columns, ordered_headers=True
)
results = controller.recorder.db.get_by_ids(ids, columns=columns, ordered_headers=True)

if derived_fields:
for entry in results:
Expand Down Expand Up @@ -438,26 +460,28 @@ async def load_traffic_file(
scope: Comma-separated list of domains to filter by during import.
Only flows matching these domains are imported.
"""
scope_list = (
[d.strip() for d in scope.split(",") if d.strip()] if scope else None
)
scope_list = [d.strip() for d in scope.split(",") if d.strip()] if scope else None

# Security: Prevent path traversal and restrict to working directory
try:
requested_path = Path(file_path).resolve()
base_dir = Path.cwd().resolve()
if not str(requested_path).startswith(str(base_dir)):
return json.dumps({
"status": "error",
"message": f"Security Error: Access denied to {file_path}. Path must be within the project directory."
})
return json.dumps(
{
"status": "error",
"message": f"Security Error: Access denied to {file_path}. Path must be within the project directory.",
}
)
except Exception as e:
return json.dumps({"status": "error", "message": f"Invalid path: {str(e)}"})

try:
stats = await asyncio.to_thread(
controller.recorder.db.import_from_file,
str(requested_path), append=append, scope=scope_list
str(requested_path),
append=append,
scope=scope_list,
)
return json.dumps(
{
Expand Down Expand Up @@ -1246,6 +1270,14 @@ async def generate_scraper_code(flow_ids: str, target_framework: str = "curl_cff
return render_scraper_code(target_framework, normalized_flows)


def _env_flag(name: str, default: bool = False) -> bool:
"""Parse a boolean env var (1/true/yes/on are truthy, case-insensitive)."""
val = os.environ.get(name)
if val is None:
return default
return val.strip().lower() in ("1", "true", "yes", "on")


def start():
"""Entry point for running the server directly."""
import argparse
Expand All @@ -1263,12 +1295,36 @@ def start():
help="Upstream proxy URL (e.g., http://user:pass@proxy:port). "
"Can also be set via MITMPROXY_UPSTREAM_PROXY env var.",
)
parser.add_argument(
"--port",
type=int,
default=int(os.environ.get("MITMPROXY_PORT", "8080")),
help="Default proxy listen port used by start_proxy and --auto-start "
"(default 8080). Can also be set via MITMPROXY_PORT env var.",
)
parser.add_argument(
"--host",
default=os.environ.get("MITMPROXY_HOST", "127.0.0.1"),
help="Default proxy listen host (default 127.0.0.1). "
"Can also be set via MITMPROXY_HOST env var.",
)
parser.add_argument(
"--auto-start",
action="store_true",
default=_env_flag("MITMPROXY_AUTO_START"),
help="Start the proxy immediately on server startup instead of waiting "
"for the start_proxy tool. Can also be set via MITMPROXY_AUTO_START env var.",
)
args, _ = parser.parse_known_args()

global controller
# Store CLI upstream proxy if provided
controller = MitmController(dump_file=args.dump_file)
controller.cli_upstream_proxy = args.upstream_proxy
# start_proxy() and --auto-start fall back to these when no port is passed,
# so a wrapper can pin a per-agent port that the browser also targets.
controller.default_port = args.port
controller.default_host = args.host
controller.auto_start = args.auto_start

mcp.run()

Expand Down