mitmproxy-mcp MCP Server

Instructions

Implementation Reference

• src/mitmproxy_mcp/server.py:133-162 (handler)

  The primary handler function for the 'list_flows' tool. It processes the session_id argument, loads flows from the dump file using a helper function, iterates over HTTP flows to create summaries (index, method, URL, status), and returns a JSON-formatted list or appropriate error messages.

  async def list_flows(arguments: dict) -> list[types.TextContent]:
      """
      Lists HTTP flows from a mitmproxy dump file.
      """
      session_id = arguments.get("session_id")
      if not session_id:
          return [types.TextContent(type="text", text="Error: Missing session_id")]
  
      try:
          flows = await get_flows_from_dump(session_id)
  
          flow_list = []
          for i, flow in enumerate(flows):
              if flow.type == "http":
                  request = flow.request
                  response = flow.response
                  flow_info = {
                      "index": i,
                      "method": request.method,
                      "url": request.url,
                      "status": response.status_code if response else None
                  }
                  flow_list.append(flow_info)
  
          return [types.TextContent(type="text", text=json.dumps(flow_list, indent=2))]
      except FileNotFoundError:
          return [types.TextContent(type="text", text="Error: Session not found")]
      except Exception as e:
          return [types.TextContent(type="text", text=f"Error reading flows: {str(e)}")]

• src/mitmproxy_mcp/server.py:38-51 (registration)

  Registration of the 'list_flows' tool in the @server.list_tools() handler, defining the tool name, description, and input JSON schema requiring 'session_id'.

  types.Tool(
      name="list_flows",
      description="Retrieves detailed HTTP request/response data including headers, content (or structure preview for large JSON), and metadata from specified flows",
      inputSchema={
          "type": "object",
          "properties": {
              "session_id": {
                  "type": "string",
                  "description": "The ID of the session to list flows from"
              }
          },
          "required": ["session_id"]
      }
  ),

• src/mitmproxy_mcp/server.py:410-411 (registration)

  Dispatch logic in the @server.call_tool() handler that routes calls to the 'list_flows' function when the tool name matches.

  if name == "list_flows":
      return await list_flows(arguments)

• src/mitmproxy_mcp/flow_utils.py:12-27 (helper)

  Supporting helper function called by the list_flows handler to load and cache mitmproxy flows from the session dump file using mitmproxy's FlowReader.

  async def get_flows_from_dump(session_id: str) -> list:
      """
      Retrieves flows from the dump file, using the cache if available.
      """
      dump_file = os.path.join(DUMP_DIR, f"{session_id}.dump")
      if not os.path.exists(dump_file):
          raise FileNotFoundError("Session not found")
  
      if session_id in FLOW_CACHE:
          return FLOW_CACHE[session_id]
      else:
          with open(dump_file, "rb") as f:
              reader = io.FlowReader(f)
              flows = list(reader.stream())
          FLOW_CACHE[session_id] = flows
          return flows