Merge pull request #9 from sleeyax/feat/load-traffic-file

feat: add `load_traffic_file` tool for importing HAR and flow files

Author: snapspecter

src/mitmproxy_mcp/core/recorder.py

@@ -1,11 +1,14 @@
 import json
+import os
 import shlex
 import sqlite3
 import sys
-from collections import deque  # added
+from collections import deque
 from typing import Any, Dict, List, Optional
+from urllib.parse import urlparse
 
 from mitmproxy import http
+from mitmproxy.io import FlowReader
 
 from .scope import ScopeManager
 from .utils import get_safe_text
@@ -285,6 +288,64 @@ def get_by_ids(self, flow_ids: List[str]) -> List[Dict[str, Any]]:
                 )
             return results
 
+    def import_from_file(
+        self,
+        file_path: str,
+        append: bool = False,
+        scope: Optional[List[str]] = None,
+    ) -> Dict[str, Any]:
+        """Import flows from a HAR or mitmproxy flow file.
+
+        Uses mitmproxy's FlowReader which auto-detects format (HAR if JSON,
+        native tnetstring otherwise).
+
+        Args:
+            file_path: Path to .har or .mitm/.flow file.
+            append: If False, clear existing traffic before import.
+            scope: Optional list of domains to filter by during import.
+
+        Returns:
+            Dict with import stats: {"imported": int, "skipped": int, "errors": int}
+        """
+        if not append:
+            self.clear()
+
+        stats = {"imported": 0, "skipped": 0, "errors": 0}
+
+        if not os.path.exists(file_path):
+            print(f"File not found: {file_path}", file=sys.stderr)
+            return stats
+
+        allowed_exts = ('.har', '.mitm', '.flow')
+        if not any(str(file_path).lower().endswith(ext) for ext in allowed_exts):
+            print(f"Unsupported file extension: {file_path}", file=sys.stderr)
+            return stats
+
+        with open(file_path, "rb") as f:
+            reader = FlowReader(f)
+            for flow in reader.stream():
+                try:
+                    if not isinstance(flow, http.HTTPFlow):
+                        stats["skipped"] += 1
+                        continue
+
+                    if scope:
+                        host = urlparse(flow.request.url).hostname or ""
+                        if not any(host == d or host.endswith("." + d) for d in scope):
+                            stats["skipped"] += 1
+                            continue
+
+                    self.save_flow(flow)
+                    stats["imported"] += 1
+                except Exception as e:
+                    stats["errors"] += 1
+                    print(
+                        f"Skipped flow during import: {e}",
+                        file=sys.stderr,
+                    )
+
+        return stats
+
     def _generate_curl(self, request: SimpleRequest) -> str:
         try:
             cmd = ["curl", "-X", request.method]

src/mitmproxy_mcp/core/server.py

@@ -249,6 +249,42 @@ async def inspect_flow(flow_id: str) -> str:
     return json.dumps(data, indent=2)
 
 
+@mcp.tool()
+async def load_traffic_file(
+    file_path: str,
+    append: bool = False,
+    scope: str = None,
+) -> str:
+    """
+    Import flows from a HAR or mitmproxy flow file into the traffic database.
+    After import, all traffic inspection tools work on the imported data.
+    No proxy needs to be running.
+    Args:
+        file_path: Path to .har or .mitm/.flow file
+        append: If True, keep existing traffic. If False (default), clear first.
+        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
+    )
+    try:
+        stats = await asyncio.to_thread(
+            controller.recorder.db.import_from_file,
+            file_path, append=append, scope=scope_list
+        )
+        return json.dumps(
+            {
+                "status": "ok",
+                "imported": stats["imported"],
+                "skipped": stats["skipped"],
+                "errors": stats["errors"],
+            }
+        )
+    except Exception as e:
+        return json.dumps({"status": "error", "message": str(e)})
+
+
 @mcp.tool()
 async def extract_from_flow(flow_id: str, json_path: str = None, css_selector: str = None) -> str:
     """